use std::collections::HashMap;
use std::io::{BufRead, BufReader};
#[cfg(test)]
use std::sync::mpsc;
use std::sync::mpsc::{Receiver, Sender};
use std::thread;
use std::time::Duration;
use ureq::http::Response;
use crate::WebSearchMode;
use crate::app::{AgentEvent, ToolStatus};
use crate::cli::{ReasoningEffort, ReasoningSummary};
use crate::providers::{
ProviderContentBlock, ProviderContinuation, ProviderError, ProviderMessage, ProviderTurn, StreamFormat,
StreamingProvider, StreamingRequest,
};
use crate::providers::{anthropic, codex, openai, opencode, umans};
use crate::tools::{self, AgentRunConfig, ToolOutput, ToolUseRequest, WriteResult, shell::ProcessResult};
use thndrs_agent::CancelToken;
use thndrs_agent::{ModelProjectionMessage, ProviderRequestAccounting, ProviderUsageComponents, ProviderUsageRule};
const PROVIDER_RETRY_POLICY: RetryPolicy = RetryPolicy::new(4, Duration::from_millis(2500));
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ProviderKind {
Fake,
Umans,
OpenCodeGo,
OpenCodeZen,
ChatGptCodex,
}
impl ProviderKind {
pub fn for_model(model: &str) -> Self {
if model.starts_with("fake-agent") {
ProviderKind::Fake
} else if opencode::is_go_model_id(model) {
ProviderKind::OpenCodeGo
} else if opencode::is_zen_model_id(model) {
ProviderKind::OpenCodeZen
} else if codex::is_model_id(model) {
ProviderKind::ChatGptCodex
} else {
ProviderKind::Umans
}
}
pub fn label(self) -> &'static str {
match self {
ProviderKind::Fake => "fake",
ProviderKind::Umans => "umans",
ProviderKind::OpenCodeGo => "opencode-go",
ProviderKind::OpenCodeZen => "opencode-zen",
ProviderKind::ChatGptCodex => "chatgpt-codex",
}
}
}
#[derive(Debug)]
enum ProviderAttemptError {
Request(ProviderError),
Stream(String),
}
impl ProviderAttemptError {
fn message<P>(&self) -> String
where
P: StreamingProvider,
{
match self {
ProviderAttemptError::Request(err) => P::request_error_message(err),
ProviderAttemptError::Stream(msg) => msg.clone(),
}
}
fn is_retryable<P>(&self) -> bool
where
P: StreamingProvider,
{
match self {
ProviderAttemptError::Request(err) => P::is_retryable_request_error(err),
ProviderAttemptError::Stream(msg) => is_retryable_stream_error(msg),
}
}
}
#[derive(Debug)]
enum MetadataLoaded<T> {
Abort,
Loaded(T),
Unavailable,
}
pub use thndrs_agent::{RetryPolicy, ToolPermissionDecision};
type ToolExecutionResult = Option<(ToolOutput, Option<WriteResult>, Option<ProcessResult>)>;
pub type ToolPermissionHook = thndrs_agent::ToolPermissionHook<AgentRunConfig>;
pub type ToolExecutionHook = thndrs_agent::ToolExecutionHook<AgentRunConfig, ToolExecutionResult>;
#[derive(Clone, Debug)]
struct ToolUseBuilder {
id: String,
name: String,
initial_input: serde_json::Value,
input_json: String,
}
impl ToolUseBuilder {
fn finish(self) -> Option<ToolUseRequest> {
let input = if self.input_json.trim().is_empty() {
self.initial_input
} else {
serde_json::from_str(&self.input_json).unwrap_or(serde_json::Value::Null)
};
let arguments = if input.is_null() {
String::from("{}")
} else {
serde_json::to_string(&input).unwrap_or_else(|_| String::from("{}"))
};
Some(ToolUseRequest::new(self.name, arguments, self.id))
}
}
#[derive(Clone, Debug)]
struct ChatToolCallBuilder {
id: String,
name: String,
arguments_json: String,
}
impl ChatToolCallBuilder {
fn finish(self) -> Option<ToolUseRequest> {
if self.name.is_empty() {
None
} else {
Some(ToolUseRequest::new(
self.name,
if self.arguments_json.trim().is_empty() { "{}".to_string() } else { self.arguments_json },
self.id,
))
}
}
}
#[derive(Debug)]
pub struct RunHandle {
pub provider: ProviderKind,
pub config: AgentRunConfig,
pub prompt: String,
pub messages: Vec<ProviderMessage>,
pub expects_write: bool,
pub steering: Option<Receiver<String>>,
pub cancel: CancelToken,
pub permission_hook: Option<ToolPermissionHook>,
pub execution_hook: Option<ToolExecutionHook>,
}
impl RunHandle {
pub fn spawn(self) -> Receiver<AgentEvent> {
let cancel = self.cancel.clone();
tracing::info!(provider = ?self.provider, "starting agent thread");
thndrs_agent::AgentRun::spawn(cancel, move |sender, cancel| self.run_agent(&sender, &cancel)).into_events()
}
pub fn fake(config: AgentRunConfig, prompt: String) -> Self {
RunHandle {
provider: ProviderKind::Fake,
config,
prompt,
messages: Vec::new(),
expects_write: false,
steering: None,
cancel: CancelToken::new(),
permission_hook: None,
execution_hook: None,
}
}
pub fn provider_with_steering(
config: AgentRunConfig, messages: Vec<ProviderMessage>, expects_write: bool, steering: Receiver<String>,
) -> Self {
let provider = ProviderKind::for_model(&config.model);
RunHandle {
provider,
config,
prompt: String::new(),
messages,
expects_write,
steering: Some(steering),
cancel: CancelToken::new(),
permission_hook: None,
execution_hook: None,
}
}
pub fn with_permission_hook(mut self, hook: ToolPermissionHook) -> Self {
self.permission_hook = Some(hook);
self
}
pub fn with_execution_hook(mut self, hook: ToolExecutionHook) -> Self {
self.execution_hook = Some(hook);
self
}
fn run_agent(&self, tx: &Sender<AgentEvent>, cancel: &CancelToken) {
if send(tx, AgentEvent::Started, cancel).is_none() {
return;
}
step();
match self.provider {
ProviderKind::Fake => self.run_fake(tx, cancel),
ProviderKind::Umans => self.run_provider::<umans::UmansClient>(tx, cancel),
ProviderKind::OpenCodeGo => self.run_provider::<opencode::OpenCodeGoClient>(tx, cancel),
ProviderKind::OpenCodeZen => self.run_provider::<opencode::zen::OpenCodeZenClient>(tx, cancel),
ProviderKind::ChatGptCodex => self.run_provider::<codex::ChatGptCodexClient>(tx, cancel),
}
}
#[expect(
clippy::cognitive_complexity,
reason = "Provider turns intentionally centralize cancellation, tool permissions, and continuation state."
)]
fn run_provider<P>(&self, tx: &Sender<AgentEvent>, cancel: &CancelToken)
where
P: StreamingProvider,
{
let provider = match P::from_env_or_dotenv(&self.config.root) {
Ok(provider) => provider,
Err(e) => {
let message = P::request_error_message(&e);
tracing::error!(error = %message, "failed to load provider client");
let _ = send(tx, AgentEvent::Failed(message), cancel);
return;
}
};
tracing::info!(
provider = provider.name(),
model = %self.config.model,
cwd = %self.config.root.display(),
messages = self.messages.len(),
max_tool_iterations = self.config.max_tool_iterations,
"starting provider agent run"
);
if send(tx, AgentEvent::Status(provider.load_status()), cancel).is_none() {
return;
}
let model_metadata = match load_provider_metadata(&provider, &self.config.model, tx, cancel) {
MetadataLoaded::Abort => return,
MetadataLoaded::Loaded(metadata) => Some(metadata),
MetadataLoaded::Unavailable => None,
};
let tool_defs = tools::runtime_tool_definitions(self.config.mcp_manager.as_deref());
let tool_schemas = tools::tool_catalog_schemas(&tool_defs);
let mut messages = if self.messages.is_empty() {
vec![ProviderMessage::user(&self.prompt)]
} else {
self.messages.clone()
};
let mut tool_budget =
thndrs_agent::ToolIterationBudget::new(self.config.max_tool_iterations, tools::MAX_TOOL_CONTINUATIONS);
let mut wrote_file = false;
let mut continuation = ProviderContinuation::default();
loop {
if cancel.is_cancelled() {
tracing::warn!(
provider = provider.name(),
"provider run cancelled before provider request"
);
let _ = send(tx, AgentEvent::Cancelled, cancel);
return;
}
match tool_budget.before_provider_request() {
thndrs_agent::ToolBudgetDecision::Continue => {}
thndrs_agent::ToolBudgetDecision::ContinueAfterBudgetMessage => {
let text = format!(
"[tool-budget]\nTool batch segment limit reached after {} total batches. Continue from the current state, avoid repeating completed work, and stop requesting tools once you can answer.",
tool_budget.total_batches()
);
tracing::warn!(
total_batches = tool_budget.total_batches(),
continuations_used = tool_budget.continuations_used(),
"continuing after tool-budget segment cap"
);
messages.push(ProviderMessage::user(&text));
if send(
tx,
AgentEvent::Status(format!(
"tool budget: auto-continue {}/{} after {} batches",
tool_budget.continuations_used(),
tools::MAX_TOOL_CONTINUATIONS,
tool_budget.total_batches()
)),
cancel,
)
.is_none()
{
return;
}
}
thndrs_agent::ToolBudgetDecision::Exhausted {
segment_iterations,
total_batches,
continuations_used,
} => {
tracing::error!(
segment_iterations,
total_batches,
continuations_used,
"tool-call budget exhausted"
);
let _ = send(
tx,
AgentEvent::Failed(format!(
"tool-call budget exhausted ({total_batches} tool batches, {continuations_used} auto-continuations, {segment_iterations} in current segment)"
)),
cancel,
);
return;
}
}
if send(
tx,
AgentEvent::Status(provider.request_status(&self.config.model)),
cancel,
)
.is_none()
{
return;
}
let max_tokens = provider.token_budget(&self.config.model, model_metadata.as_ref());
let request = ProviderTurnRequest {
provider: &provider,
model: &self.config.model,
messages: &messages,
max_tokens,
reasoning_effort: self.config.reasoning_effort,
reasoning_summary: self.config.reasoning_summary,
tool_schemas: &tool_schemas,
continuation: &continuation,
turn_id: self.config.accounting_turn_id.as_deref().unwrap_or("turn_unknown"),
context: &self.config.accounting_context,
};
let Some(mut turn) = request_provider_turn_with_retries(&request, tool_budget.total_batches(), tx, cancel)
else {
return;
};
if matches!(self.provider, ProviderKind::ChatGptCodex | ProviderKind::OpenCodeZen)
&& matches!(
provider.stream_format(&self.config.model),
Ok(StreamFormat::ChatGptCodexResponses)
)
{
codex::record_response_items(&mut continuation, &messages, std::mem::take(&mut turn.response_items));
}
tracing::info!(
text_chars = turn.assistant_text.chars().count(),
tool_calls = turn.tool_requests.len(),
"provider turn completed"
);
if turn.tool_requests.is_empty() {
if turn.assistant_text.is_empty() && turn.stop_reason.as_deref() == Some("max_tokens") {
let _ = send(
tx,
AgentEvent::Failed(format!(
"provider stopped at max_tokens ({max_tokens}) before producing assistant text"
)),
cancel,
);
return;
}
if self.expects_write && !wrote_file {
let _ = send(
tx,
AgentEvent::Failed(String::from(
"model stopped without writing a file for an edit-like request",
)),
cancel,
);
return;
}
if append_steering_messages(&mut messages, self) {
tracing::info!(
provider = provider.name(),
"continuing provider run with queued steering messages"
);
continue;
}
let _ = send(tx, AgentEvent::Finished, cancel);
return;
}
tool_budget.record_tool_batch();
let mut assistant_blocks = Vec::new();
if !turn.assistant_text.is_empty() {
assistant_blocks.push(ProviderContentBlock::Text { text: turn.assistant_text });
}
let mut tool_results: Vec<ProviderMessage> = Vec::new();
let mut response_tool_outputs = Vec::new();
for req in &turn.tool_requests {
if cancel.is_cancelled() {
tracing::warn!(provider = provider.name(), tool = %req.name, tool_id = %req.tool_use_id, "provider run cancelled before tool dispatch");
let _ = send(tx, AgentEvent::Cancelled, cancel);
return;
}
let tool_id = req.tool_use_id.clone();
tracing::info!(tool = %req.name, tool_id = %tool_id, "dispatching tool request");
if send(
tx,
AgentEvent::ToolStarted {
id: tool_id.clone(),
name: req.name.clone(),
arguments: req.arguments.clone(),
},
cancel,
)
.is_none()
{
return;
}
let (output, write_result, shell_result) = match approve_tool_request(req, self, cancel) {
ToolPermissionDecision::Allow => dispatch_tool_request(req, self, cancel),
ToolPermissionDecision::Reject => (
ToolOutput::failed(&req.name, String::from("tool call rejected by ACP client")),
None,
None,
),
ToolPermissionDecision::Cancelled => {
let _ = send(tx, AgentEvent::Cancelled, cancel);
return;
}
};
let status = output.status;
let display_output = output.display_lines();
let model_output = output.model_lines();
if write_result.is_some() && status == ToolStatus::Ok {
wrote_file = true;
}
tracing::info!(tool = %req.name, tool_id = %tool_id, status = ?status, "tool request finished");
if send(
tx,
AgentEvent::ToolFinished {
id: tool_id.clone(),
output: display_output.clone(),
status,
write_result,
shell_result: shell_result.map(Box::new),
},
cancel,
)
.is_none()
{
return;
}
let input: serde_json::Value = serde_json::from_str(&req.arguments).unwrap_or(serde_json::Value::Null);
assistant_blocks.push(ProviderContentBlock::ToolUse {
id: tool_id.clone(),
name: req.name.clone(),
input,
});
let result_content =
if model_output.is_empty() { "(no output)".to_string() } else { model_output.join("\n") };
let is_error = status == ToolStatus::Failed;
tool_results.push(ProviderMessage::tool_result(&tool_id, &result_content, is_error));
response_tool_outputs.push((tool_id, result_content));
}
messages.push(ProviderMessage::assistant_blocks(assistant_blocks));
messages.extend(tool_results);
if matches!(self.provider, ProviderKind::ChatGptCodex | ProviderKind::OpenCodeZen)
&& matches!(
provider.stream_format(&self.config.model),
Ok(StreamFormat::ChatGptCodexResponses)
)
{
for (call_id, output) in response_tool_outputs {
codex::record_tool_output(&mut continuation, &call_id, &output, messages.len());
}
}
append_steering_messages(&mut messages, self);
}
}
fn run_fake(&self, tx: &Sender<AgentEvent>, cancel: &CancelToken) {
use AgentEvent::*;
match send(tx, ReasoningDelta(String::from("Let me think about this... ")), cancel) {
None => return,
Some(_) => step(),
}
match send(
tx,
ReasoningDelta(String::from("The repo is a Rust terminal coding harness.")),
cancel,
) {
None => return,
Some(_) => step(),
}
if self.config.model == "fake-agent-slow" {
for _ in 0..200 {
if cancel.is_cancelled() {
let _ = send(tx, Cancelled, cancel);
return;
}
step();
}
}
if self.config.search_mode != WebSearchMode::None {
let search_req = ToolUseRequest::new(
String::from("web_search"),
serde_json::json!({ "query": "rust terminal coding harness" }).to_string(),
String::from("search-0"),
);
let search_id = String::from("search-0");
match send(
tx,
ToolStarted {
id: search_id.clone(),
name: search_req.name.clone(),
arguments: search_req.arguments.clone(),
},
cancel,
) {
None => return,
Some(_) => step(),
}
let search_config = self.config.search_config();
let (search_output, _, _) =
tools::dispatch_full_with_search(&search_req, &self.config.root, &search_config);
let search_status = search_output.status;
let search_display_output = search_output.display_lines();
match send(
tx,
ToolFinished {
id: search_id,
output: search_display_output,
status: search_status,
write_result: None,
shell_result: None,
},
cancel,
) {
None => return,
Some(_) => {
step();
}
}
}
let tool_req = ToolUseRequest::new(
String::from("read_file_range"),
serde_json::json!({ "path": "Cargo.toml", "start_line": 1, "end_line": 5 }).to_string(),
String::from("0"),
);
let tool_id = String::from("0");
match send(
tx,
ToolStarted { id: tool_id.clone(), name: tool_req.name.clone(), arguments: tool_req.arguments.clone() },
cancel,
) {
None => return,
Some(_) => step(),
}
let (output, _, _) = tools::dispatch_full(&tool_req, &self.config.root);
let status = output.status;
let display_output = output.display_lines();
match send(
tx,
ToolFinished { id: tool_id, output: display_output, status, write_result: None, shell_result: None },
cancel,
) {
None => return,
Some(_) => step(),
}
if self.config.model == "fake-agent-shell" {
let shell_req = ToolUseRequest::new(
String::from("run_shell"),
serde_json::json!({ "program": "printf", "args": ["acp-permission-smoke\n"] }).to_string(),
String::from("shell-0"),
);
let shell_id = String::from("shell-0");
match send(
tx,
ToolStarted {
id: shell_id.clone(),
name: shell_req.name.clone(),
arguments: shell_req.arguments.clone(),
},
cancel,
) {
None => return,
Some(_) => step(),
}
let (shell_output, write_result, shell_result) = match approve_tool_request(&shell_req, self, cancel) {
ToolPermissionDecision::Allow => dispatch_tool_request(&shell_req, self, cancel),
ToolPermissionDecision::Reject => (
ToolOutput::failed(&shell_req.name, String::from("tool call rejected by ACP client")),
None,
None,
),
ToolPermissionDecision::Cancelled => {
let _ = send(tx, AgentEvent::Cancelled, cancel);
return;
}
};
let shell_status = shell_output.status;
let shell_display_output = shell_output.display_lines();
match send(
tx,
ToolFinished {
id: shell_id,
output: shell_display_output,
status: shell_status,
write_result,
shell_result: shell_result.map(Box::new),
},
cancel,
) {
None => return,
Some(_) => step(),
}
}
match send(tx, AssistantDelta(String::from("This is a ")), cancel) {
None => return,
Some(_) => step(),
}
match send(tx, AssistantDelta(String::from("fake streaming response.")), cancel) {
None => return,
Some(_) => step(),
}
let _ = tx.send(Finished);
}
}
#[derive(Default)]
struct AnthropicStreamState {
tool_blocks: HashMap<usize, ToolUseBuilder>,
tool_requests: Vec<ToolUseRequest>,
assistant_text: String,
stop_reason: Option<String>,
provider_content_blocks: Vec<String>,
usage: ProviderUsageComponents,
}
#[cfg(test)]
impl AnthropicStreamState {
fn collect(
&mut self, event_type: &str, data: &str, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Result<(), String> {
collect_anthropic_event(event_type, data, self, tx, cancel)
}
}
struct ProviderTurnRequest<'a, P>
where
P: StreamingProvider,
{
provider: &'a P,
model: &'a str,
messages: &'a [ProviderMessage],
max_tokens: u32,
reasoning_effort: ReasoningEffort,
reasoning_summary: ReasoningSummary,
tool_schemas: &'a serde_json::Value,
continuation: &'a ProviderContinuation,
turn_id: &'a str,
context: &'a [thndrs_agent::ContextItemSnapshot],
}
pub fn prompt_expects_workspace_write(prompt: &str) -> bool {
let lower = prompt.to_ascii_lowercase();
let fileish = lower.contains(".md")
|| lower.contains(".rs")
|| lower.contains(".toml")
|| lower.contains(".json")
|| lower.contains(".yaml")
|| lower.contains(".yml")
|| lower.contains("file")
|| lower.contains("todo");
let action = [
"add",
"change",
"document",
"edit",
"fix",
"modify",
"remove",
"replace",
"rewrite",
"summarize",
"update",
"write",
]
.iter()
.any(|word| {
lower
.split(|c: char| !c.is_ascii_alphanumeric())
.any(|part| part == *word)
});
fileish && action
}
fn is_retryable_stream_error(message: &str) -> bool {
let lower = message.to_ascii_lowercase();
if lower.contains("cancel")
|| lower.contains("aborted")
|| lower.contains("max_tokens")
|| lower.contains("without writing")
|| lower.contains("provider returned only provider-side content blocks")
{
return false;
}
[
"429",
"500",
"502",
"503",
"504",
"overloaded",
"rate limit",
"server error",
"service unavailable",
"stream read error",
"stream ended without",
"connection",
"timed out",
"timeout",
"provider error",
]
.iter()
.any(|needle| lower.contains(needle))
}
fn sleep_with_cancel(delay: Duration, tx: &Sender<AgentEvent>, cancel: &CancelToken) -> bool {
let mut slept = Duration::ZERO;
let tick = Duration::from_millis(100);
while slept < delay {
if cancel.is_cancelled() {
let _ = tx.send(AgentEvent::Cancelled);
return false;
}
let remaining = delay.saturating_sub(slept);
let nap = remaining.min(tick);
thread::sleep(nap);
slept += nap;
}
true
}
fn dispatch_tool_request(
request: &ToolUseRequest, handle: &RunHandle, cancel: &CancelToken,
) -> (ToolOutput, Option<WriteResult>, Option<ProcessResult>) {
if let Some(hook) = &handle.execution_hook
&& let Some(output) = hook.execute(request, &handle.config, cancel)
{
return output;
}
let search_config = handle.config.search_config();
tools::dispatch_runtime_full_with_cancel_and_search_and_registry(
request,
&handle.config.root,
handle.config.mcp_manager.as_deref(),
cancel,
&search_config,
handle.config.process_registry.as_ref(),
)
}
fn approve_tool_request(request: &ToolUseRequest, handle: &RunHandle, cancel: &CancelToken) -> ToolPermissionDecision {
if !requires_runtime_permission(&request.name) {
return ToolPermissionDecision::Allow;
}
match &handle.permission_hook {
Some(hook) => hook.decide(request, &handle.config, cancel),
None => ToolPermissionDecision::Allow,
}
}
fn requires_runtime_permission(tool_name: &str) -> bool {
matches!(tool_name, "create_file" | "replace_range" | "write_patch" | "run_shell")
}
fn load_provider_metadata<P>(
provider: &P, model: &str, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> MetadataLoaded<P::Metadata>
where
P: StreamingProvider,
{
match provider.load_metadata() {
Ok(models) => {
tracing::info!("loaded provider model metadata");
if let Some(event) = provider.metadata_loaded_event(&models)
&& send(tx, event, cancel).is_none()
{
return MetadataLoaded::Abort;
}
if let Some(status) = provider.metadata_status(model, &models)
&& send(tx, AgentEvent::Status(status), cancel).is_none()
{
return MetadataLoaded::Abort;
}
MetadataLoaded::Loaded(models)
}
Err(e) => {
let message = P::request_error_message(&e);
if e.is_credential_rejected() {
tracing::warn!(error = %message, "provider rejected credentials while loading model metadata");
let _ = send(tx, AgentEvent::Failed(message), cancel);
return MetadataLoaded::Abort;
}
tracing::warn!(error = %message, "failed to load provider model metadata; using fallback token budget");
match send(
tx,
AgentEvent::Status(String::from(
"provider: model metadata unavailable; using fallback token budget",
)),
cancel,
) {
None => MetadataLoaded::Abort,
Some(_) => MetadataLoaded::Unavailable,
}
}
}
}
fn request_provider_turn_with_retries<P>(
request: &ProviderTurnRequest<'_, P>, iteration: usize, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Option<ProviderTurn>
where
P: StreamingProvider,
{
let mut retry_attempt = 0;
loop {
tracing::info!(
iteration,
messages = request.messages.len(),
retry_attempt,
"requesting provider turn"
);
let attempt_result = provider_request_attempt(request, iteration, retry_attempt + 1, tx, cancel);
match attempt_result {
Ok(turn) => return Some(turn),
Err(error) if error.is_retryable::<P>() && retry_attempt < PROVIDER_RETRY_POLICY.max_retries => {
retry_attempt += 1;
if !send_retry_event(request.provider, error.message::<P>(), retry_attempt, tx, cancel) {
return None;
}
}
Err(error) => {
let message = error.message::<P>();
tracing::error!(provider = request.provider.name(), error = %message, "provider attempt failed");
let _ = send(tx, AgentEvent::Failed(message), cancel);
return None;
}
}
}
}
fn provider_request_attempt<P>(
request: &ProviderTurnRequest<'_, P>, iteration: usize, attempt: u32, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Result<ProviderTurn, ProviderAttemptError>
where
P: StreamingProvider,
{
let provider_request = StreamingRequest {
max_tokens: request.max_tokens,
reasoning_effort: request.reasoning_effort,
reasoning_summary: request.reasoning_summary,
tools: request.tool_schemas,
continuation: request.continuation,
};
let serialized_body = request
.provider
.serialized_request_body(request.model, request.messages, &provider_request)
.map_err(ProviderAttemptError::Request)?;
match request
.provider
.send_streaming_request(request.model, request.messages, &provider_request)
{
Ok(response) => {
match send(
tx,
AgentEvent::Status(format!("provider: connected HTTP {}", response.status().as_u16())),
cancel,
) {
None => Err(ProviderAttemptError::Stream("cancelled".to_string())),
Some(_) => stream_provider_response(
request.provider,
request.model,
response,
tx,
cancel,
request.max_tokens,
)
.map_err(ProviderAttemptError::Stream)
.and_then(|mut turn| {
let stream_format = request
.provider
.stream_format(request.model)
.map_err(ProviderAttemptError::Request)?;
let provider_usage = turn.usage.take().and_then(|components| {
if components.input_tokens.is_none()
&& components.output_tokens.is_none()
&& components.cache_read_input_tokens.is_none()
&& components.cache_creation_input_tokens.is_none()
&& components.reasoning_tokens.is_none()
{
None
} else {
Some(
components
.normalize(request.provider.name(), usage_rule_for_stream_format(stream_format)),
)
}
});
let mut accounting = ProviderRequestAccounting::from_serialized_request(
request.turn_id,
format!("{}:request:{iteration}", request.turn_id),
attempt,
request.provider.name(),
request.model,
&serialized_body,
request.context.to_vec(),
);
accounting = accounting.with_model_projection(
request
.messages
.iter()
.map(|message| ModelProjectionMessage {
role: message.role.clone(),
content: match &message.content {
crate::providers::ProviderMessageContent::Text(content) => content.clone(),
crate::providers::ProviderMessageContent::Blocks(blocks) => {
serde_json::to_string(blocks)
.unwrap_or_else(|_| String::from("[unserializable blocks]"))
}
},
})
.collect(),
);
accounting.provider_usage = provider_usage;
if send(tx, AgentEvent::RequestAccounting(Box::new(accounting)), cancel).is_none() {
return Err(ProviderAttemptError::Stream("cancelled".to_string()));
}
Ok(turn)
}),
}
}
Err(e) => Err(ProviderAttemptError::Request(e)),
}
}
fn usage_rule_for_stream_format(format: StreamFormat) -> ProviderUsageRule {
match format {
StreamFormat::AnthropicMessages => ProviderUsageRule::AnthropicMessages,
StreamFormat::OpenAiChat => ProviderUsageRule::OpenAiChat,
StreamFormat::ChatGptCodexResponses => ProviderUsageRule::OpenAiResponses,
}
}
fn send_retry_event<P>(
provider: &P, message: String, retry_attempt: u32, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> bool
where
P: StreamingProvider,
{
let delay = PROVIDER_RETRY_POLICY.delay_for_attempt(retry_attempt);
tracing::warn!(
provider = provider.name(),
attempt = retry_attempt,
max_retries = PROVIDER_RETRY_POLICY.max_retries,
delay_ms = delay.as_millis(),
error = %message,
"retrying provider attempt"
);
match send(
tx,
AgentEvent::Retrying {
attempt: retry_attempt,
max_attempts: PROVIDER_RETRY_POLICY.max_retries,
delay_ms: delay.as_millis() as u64,
error: message,
},
cancel,
) {
None => false,
Some(_) => sleep_with_cancel(delay, tx, cancel),
}
}
fn append_steering_messages(messages: &mut Vec<ProviderMessage>, handle: &RunHandle) -> bool {
match handle.steering.as_ref() {
Some(rx) => {
let mut appended = false;
while let Ok(text) = rx.try_recv() {
let trimmed = text.trim();
if trimmed.is_empty() {
continue;
}
messages.push(ProviderMessage::user(&format!("[steering]\n{trimmed}")));
appended = true;
}
if appended {
tracing::debug!(messages = messages.len(), "appended steering messages");
}
appended
}
None => false,
}
}
fn stream_anthropic_response(
resp: Response<ureq::Body>, tx: &Sender<AgentEvent>, cancel: &CancelToken, max_tokens: u32,
) -> Result<ProviderTurn, String> {
let reader = BufReader::new(resp.into_body().into_reader());
let mut buffer = String::new();
let mut state = AnthropicStreamState::default();
let mut event_count = 0usize;
let mut saw_response = false;
tracing::info!("reading Anthropic-compatible SSE stream");
for line_result in reader.lines() {
if cancel.is_cancelled() {
tracing::warn!("cancelled while reading Anthropic-compatible SSE stream");
return Err("cancelled".to_string());
}
match line_result {
Ok(line) => {
buffer.push_str(&line);
buffer.push('\n');
if line.is_empty() {
let events = anthropic::parse_sse_chunk(&buffer);
buffer.clear();
for (event_type, data) in events {
event_count += 1;
log_sse_event(event_count, &event_type, &data);
if !saw_response {
saw_response = true;
if send(tx, AgentEvent::Status(String::from("provider: receiving SSE")), cancel).is_none() {
return Err("cancelled".to_string());
}
}
collect_anthropic_event(&event_type, &data, &mut state, tx, cancel)?;
}
}
}
Err(e) => {
tracing::error!(error = %e, "failed reading Anthropic-compatible SSE stream");
return Err(format!("stream read error: {e}"));
}
}
}
if !buffer.is_empty() {
let events = anthropic::parse_sse_chunk(&buffer);
for (event_type, data) in events {
event_count += 1;
log_sse_event(event_count, &event_type, &data);
if !saw_response {
saw_response = true;
if send(tx, AgentEvent::Status(String::from("provider: receiving SSE")), cancel).is_none() {
return Err("cancelled".to_string());
}
}
collect_anthropic_event(&event_type, &data, &mut state, tx, cancel)?;
}
}
for (_, block) in state.tool_blocks {
if let Some(req) = block.finish() {
state.tool_requests.push(req);
}
}
if state.assistant_text.is_empty() && state.tool_requests.is_empty() {
tracing::error!(
event_count,
"Anthropic-compatible stream ended without assistant text or tool calls"
);
if state.stop_reason.as_deref() == Some("max_tokens") {
return Err(format!(
"provider stopped at max_tokens ({max_tokens}) before producing assistant text"
));
}
if !state.provider_content_blocks.is_empty() {
let blocks = state.provider_content_blocks.join(", ");
return Err(format!(
"provider returned only provider-side content blocks ({blocks}) and no assistant text or tool calls; retry with --websearch none"
));
}
return Err(format!(
"provider stream ended without assistant text or tool calls ({event_count} SSE events)"
));
}
tracing::info!(
event_count,
text_chars = state.assistant_text.chars().count(),
tool_calls = state.tool_requests.len(),
"finished reading Anthropic-compatible SSE stream"
);
let _ = send(
tx,
AgentEvent::Status(format!(
"provider: stream ended ({event_count} SSE events, {} text chars, {} tool calls)",
state.assistant_text.chars().count(),
state.tool_requests.len()
)),
cancel,
);
Ok(ProviderTurn {
tool_requests: state.tool_requests,
assistant_text: state.assistant_text,
stop_reason: state.stop_reason,
response_items: Vec::new(),
usage: Some(state.usage),
})
}
fn stream_provider_response<P: StreamingProvider>(
provider: &P, model: &str, resp: Response<ureq::Body>, tx: &Sender<AgentEvent>, cancel: &CancelToken,
max_tokens: u32,
) -> Result<ProviderTurn, String> {
match provider
.stream_format(model)
.map_err(|e| P::request_error_message(&e))?
{
StreamFormat::OpenAiChat => stream_openai_chat_response(resp, tx, cancel, max_tokens),
StreamFormat::AnthropicMessages => stream_anthropic_response(resp, tx, cancel, max_tokens),
StreamFormat::ChatGptCodexResponses => stream_chatgpt_codex_response(resp, tx, cancel, max_tokens),
}
}
fn stream_openai_chat_response(
resp: Response<ureq::Body>, tx: &Sender<AgentEvent>, cancel: &CancelToken, max_tokens: u32,
) -> Result<ProviderTurn, String> {
let reader = BufReader::new(resp.into_body().into_reader());
let mut assistant_text = String::new();
let mut tool_blocks: HashMap<usize, ChatToolCallBuilder> = HashMap::new();
let mut tool_requests = Vec::new();
let mut event_count = 0usize;
let mut saw_response = false;
let mut stop_reason = None;
let mut usage = ProviderUsageComponents::default();
tracing::info!("reading OpenAI-compatible chat-completions SSE stream");
for line_result in reader.lines() {
if cancel.is_cancelled() {
tracing::warn!("cancelled while reading OpenAI-compatible SSE stream");
return Err("cancelled".to_string());
}
let line = line_result.map_err(|e| {
tracing::error!(error = %e, "failed reading OpenAI-compatible SSE stream");
format!("stream read error: {e}")
})?;
if !line.starts_with("data:") {
continue;
}
if !saw_response {
saw_response = true;
if send(tx, AgentEvent::Status(String::from("provider: receiving SSE")), cancel).is_none() {
return Err("cancelled".to_string());
}
}
for data in openai::parse_chat_sse_chunk(&(line + "\n")) {
event_count += 1;
let events = openai::parse_chat_sse_event(&data);
for event in events {
match &event {
openai::ChatSseEvent::Usage { input_tokens, output_tokens } => {
usage.merge_snapshot(&ProviderUsageComponents::new(*input_tokens, *output_tokens));
}
openai::ChatSseEvent::UsageComponents(components) => usage.merge_snapshot(components),
_ => {}
}
collect_openai_chat_event(
event,
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
tx,
cancel,
)?;
}
}
}
for (_, block) in tool_blocks {
if let Some(req) = block.finish() {
tool_requests.push(req);
}
}
if assistant_text.is_empty() && tool_requests.is_empty() {
tracing::error!(
event_count,
"OpenAI-compatible stream ended without assistant text or tool calls"
);
return match stop_reason.as_deref() {
Some("length") => Err(format!(
"provider stopped at max_tokens ({max_tokens}) before producing assistant text"
)),
_ => Err(format!(
"provider stream ended without assistant text or tool calls ({event_count} SSE events)"
)),
};
}
tracing::info!(
event_count,
text_chars = assistant_text.chars().count(),
tool_calls = tool_requests.len(),
"finished reading OpenAI-compatible SSE stream"
);
let _ = send(
tx,
AgentEvent::Status(format!(
"provider: stream ended ({event_count} SSE events, {} text chars, {} tool calls)",
assistant_text.chars().count(),
tool_requests.len()
)),
cancel,
);
Ok(ProviderTurn { tool_requests, assistant_text, stop_reason, response_items: Vec::new(), usage: Some(usage) })
}
fn collect_openai_chat_event(
event: openai::ChatSseEvent, tool_blocks: &mut HashMap<usize, ChatToolCallBuilder>,
tool_requests: &mut Vec<ToolUseRequest>, assistant_text: &mut String, stop_reason: &mut Option<String>,
tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Result<(), String> {
match event {
openai::ChatSseEvent::TextDelta(text) => {
assistant_text.push_str(&text);
if send(tx, AgentEvent::AssistantDelta(text), cancel).is_none() {
return Err("cancelled".to_string());
}
}
openai::ChatSseEvent::ReasoningDelta(text) => {
if send(tx, AgentEvent::ReasoningDelta(text), cancel).is_none() {
return Err("cancelled".to_string());
}
}
openai::ChatSseEvent::ToolCallStart { index, id, name } => {
tool_blocks.insert(index, ChatToolCallBuilder { id, name, arguments_json: String::new() });
}
openai::ChatSseEvent::ToolCallArgumentsDelta { index, arguments } => {
let block = tool_blocks.entry(index).or_insert_with(|| ChatToolCallBuilder {
id: format!("call_{index}"),
name: String::new(),
arguments_json: String::new(),
});
block.arguments_json.push_str(&arguments);
}
openai::ChatSseEvent::FinishReason(reason) => {
*stop_reason = Some(reason.clone());
if reason == "tool_calls" {
let finished = std::mem::take(tool_blocks);
for (_, block) in finished {
if let Some(req) = block.finish() {
tool_requests.push(req);
}
}
}
}
openai::ChatSseEvent::ResponseStatus(status) => {
if matches!(status.as_str(), "failed" | "cancelled" | "canceled") {
return Err(format!("provider stream status: {status}"));
}
if send(tx, AgentEvent::Status(format!("provider: status {status}")), cancel).is_none() {
return Err("cancelled".to_string());
}
}
openai::ChatSseEvent::Error(message) => {
tracing::error!(error = %message, "provider emitted SSE error");
return Err(format!("provider error: {message}"));
}
openai::ChatSseEvent::Usage { .. } => {}
openai::ChatSseEvent::UsageComponents(_) => {}
openai::ChatSseEvent::Malformed(payload) => {
return Err(format!("malformed provider stream payload: {payload}"));
}
openai::ChatSseEvent::Done | openai::ChatSseEvent::Other => {}
}
Ok(())
}
fn stream_chatgpt_codex_response(
resp: Response<ureq::Body>, tx: &Sender<AgentEvent>, cancel: &CancelToken, max_tokens: u32,
) -> Result<ProviderTurn, String> {
let reader = BufReader::new(resp.into_body().into_reader());
let mut assistant_text = String::new();
let mut tool_blocks: HashMap<String, ChatToolCallBuilder> = HashMap::new();
let mut tool_requests = Vec::new();
let mut event_count = 0usize;
let mut saw_response = false;
let mut stop_reason = None;
let mut response_items = Vec::new();
let mut usage = ProviderUsageComponents::default();
tracing::info!("reading ChatGPT Codex Responses SSE stream");
for line_result in reader.lines() {
if cancel.is_cancelled() {
tracing::warn!("cancelled while reading ChatGPT Codex SSE stream");
return Err("cancelled".to_string());
}
let line = line_result.map_err(|e| {
tracing::error!(error = %e, "failed reading ChatGPT Codex SSE stream");
format!("stream read error: {e}")
})?;
if !line.starts_with("data:") {
continue;
}
if !saw_response {
saw_response = true;
if send(
tx,
AgentEvent::Status(String::from("provider: receiving ChatGPT Codex SSE")),
cancel,
)
.is_none()
{
return Err("cancelled".to_string());
}
}
for data in codex::parse_responses_sse_chunk(&(line + "\n")) {
event_count += 1;
for event in codex::parse_responses_sse_event(&data) {
match &event {
codex::ResponsesSseEvent::Usage { input_tokens, output_tokens } => {
usage.merge_snapshot(&ProviderUsageComponents::new(*input_tokens, *output_tokens));
}
codex::ResponsesSseEvent::UsageComponents(components) => usage.merge_snapshot(components),
_ => {}
}
if let codex::ResponsesSseEvent::OutputItem(item) = &event {
response_items.push(item.clone());
}
collect_chatgpt_codex_event(
event,
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
tx,
cancel,
)?;
}
}
}
for (_, block) in tool_blocks {
if let Some(req) = block.finish() {
tool_requests.push(req);
}
}
if assistant_text.is_empty() && tool_requests.is_empty() {
tracing::error!(
event_count,
"ChatGPT Codex stream ended without assistant text or tool calls"
);
return match stop_reason.as_deref() {
Some("incomplete" | "length") => Err(format!(
"provider stopped at max_tokens ({max_tokens}) before producing assistant text"
)),
_ => Err(format!(
"provider stream ended without assistant text or tool calls ({event_count} SSE events)"
)),
};
}
tracing::info!(
event_count,
text_chars = assistant_text.chars().count(),
tool_calls = tool_requests.len(),
"finished reading ChatGPT Codex SSE stream"
);
let _ = send(
tx,
AgentEvent::Status(format!(
"provider: stream ended ({event_count} SSE events, {} text chars, {} tool calls)",
assistant_text.chars().count(),
tool_requests.len()
)),
cancel,
);
Ok(ProviderTurn { tool_requests, assistant_text, stop_reason, response_items, usage: Some(usage) })
}
fn collect_chatgpt_codex_event(
event: codex::ResponsesSseEvent, tool_blocks: &mut HashMap<String, ChatToolCallBuilder>,
tool_requests: &mut Vec<ToolUseRequest>, assistant_text: &mut String, stop_reason: &mut Option<String>,
tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Result<(), String> {
match event {
codex::ResponsesSseEvent::TextDelta(text) => {
assistant_text.push_str(&text);
if send(tx, AgentEvent::AssistantDelta(text), cancel).is_none() {
return Err("cancelled".to_string());
}
}
codex::ResponsesSseEvent::ReasoningDelta(text) => {
if send(tx, AgentEvent::ReasoningDelta(text), cancel).is_none() {
return Err("cancelled".to_string());
}
}
codex::ResponsesSseEvent::ToolCallStart { id, call_id, name } => {
tool_blocks.insert(
id,
ChatToolCallBuilder { id: call_id, name, arguments_json: String::new() },
);
}
codex::ResponsesSseEvent::ToolCallArgumentsDelta { id, arguments } => {
let block = tool_blocks.entry(id.clone()).or_insert_with(|| ChatToolCallBuilder {
id,
name: String::new(),
arguments_json: String::new(),
});
block.arguments_json.push_str(&arguments);
}
codex::ResponsesSseEvent::ToolCallDone { id, call_id, name, arguments } => {
let remove_id = id.clone();
let block = tool_blocks.entry(id.clone()).or_insert_with(|| ChatToolCallBuilder {
id: id.clone(),
name: String::new(),
arguments_json: String::new(),
});
if let Some(call_id) = call_id {
block.id = call_id;
}
if !name.is_empty() {
block.name = name;
}
if !arguments.is_empty() {
block.arguments_json = arguments;
}
if let Some(block) = tool_blocks.remove(&remove_id)
&& let Some(req) = block.finish()
{
tool_requests.push(req);
}
}
codex::ResponsesSseEvent::ResponseStatus(status) => {
match status.as_str() {
"completed" => *stop_reason = Some(status.clone()),
"failed" | "incomplete" | "cancelled" | "canceled" => {
return Err(format!("provider stream status: {status}"));
}
_ => {}
}
if send(tx, AgentEvent::Status(format!("provider: status {status}")), cancel).is_none() {
return Err("cancelled".to_string());
}
}
codex::ResponsesSseEvent::Error(message) => {
tracing::error!(error = %message, "ChatGPT Codex emitted SSE error");
return Err(format!("provider error: {message}"));
}
codex::ResponsesSseEvent::Usage { .. } => {}
codex::ResponsesSseEvent::UsageComponents(_) => {}
codex::ResponsesSseEvent::OutputItem(_) => {}
codex::ResponsesSseEvent::Malformed(payload) => {
return Err(format!("malformed provider stream payload: {payload}"));
}
codex::ResponsesSseEvent::Done | codex::ResponsesSseEvent::Other => {}
}
Ok(())
}
fn log_sse_event(seq: usize, event_type: &str, data: &str) {
let (content_type, delta_type, stop_reason) = summarize_sse_data(data);
tracing::info!(
seq,
event_type,
content_type = content_type.as_deref().unwrap_or(""),
delta_type = delta_type.as_deref().unwrap_or(""),
stop_reason = stop_reason.as_deref().unwrap_or(""),
"received SSE event"
);
}
fn summarize_sse_data(data: &str) -> (Option<String>, Option<String>, Option<String>) {
let v: serde_json::Value = serde_json::from_str(data).unwrap_or(serde_json::Value::Null);
let content_type = v
.get("content_block")
.and_then(|cb| cb.get("type"))
.and_then(|t| t.as_str())
.map(|s| s.to_string());
let delta_type = v
.get("delta")
.and_then(|d| d.get("type"))
.and_then(|t| t.as_str())
.map(|s| s.to_string());
let stop_reason = v
.get("delta")
.and_then(|d| d.get("stop_reason"))
.and_then(|s| s.as_str())
.map(|s| s.to_string());
(content_type, delta_type, stop_reason)
}
fn collect_anthropic_event(
event_type: &str, data: &str, state: &mut AnthropicStreamState, tx: &Sender<AgentEvent>, cancel: &CancelToken,
) -> Result<(), String> {
let sse_event = anthropic::parse_sse_event(event_type, data);
if let Some(usage) = extract_usage(data) {
state.usage.merge_snapshot(&usage);
}
if event_type == "content_block_start"
&& let Some((index, block)) = extract_tool_use_start(data)
{
state.tool_blocks.insert(index, block);
}
if event_type == "content_block_start" {
collect_content_block_start_text(
data,
&mut state.assistant_text,
&mut state.provider_content_blocks,
tx,
cancel,
)?;
}
match &sse_event {
anthropic::SseEvent::TextDelta(text) => state.assistant_text.push_str(text),
anthropic::SseEvent::InputJsonDelta { index, partial_json } => {
if let Some(block) = state.tool_blocks.get_mut(index) {
block.input_json.push_str(partial_json);
}
}
anthropic::SseEvent::ContentBlockStop { index } => {
if let Some(index) = index
&& let Some(block) = state.tool_blocks.remove(index)
&& let Some(req) = block.finish()
{
state.tool_requests.push(req);
}
}
anthropic::SseEvent::MessageDelta { stop_reason: Some(reason) } => {
state.stop_reason = Some(reason.clone());
tracing::info!(stop_reason = %reason, "provider message stop reason");
}
anthropic::SseEvent::Error(msg) => {
tracing::error!(error = %msg, "provider emitted SSE error");
return Err(format!("provider error: {msg}"));
}
_ => {}
}
if let Some(agent_event) = anthropic::sse_to_agent_event(&sse_event)
&& send(tx, agent_event, cancel).is_none()
{
return Err("cancelled".to_string());
}
Ok(())
}
fn extract_usage(data: &str) -> Option<ProviderUsageComponents> {
let v: serde_json::Value = serde_json::from_str(data).ok()?;
let usage = v
.get("usage")
.or_else(|| v.get("message").and_then(|m| m.get("usage")))
.or_else(|| v.get("delta").and_then(|d| d.get("usage")))?;
let input_tokens = usage.get("input_tokens").and_then(|t| t.as_u64());
let output_tokens = usage.get("output_tokens").and_then(|t| t.as_u64());
let cache_read_input_tokens = usage.get("cache_read_input_tokens").and_then(|value| value.as_u64());
let cache_creation_input_tokens = usage
.get("cache_creation_input_tokens")
.and_then(|value| value.as_u64());
if input_tokens.is_none()
&& output_tokens.is_none()
&& cache_read_input_tokens.is_none()
&& cache_creation_input_tokens.is_none()
{
None
} else {
Some(ProviderUsageComponents {
input_tokens,
output_tokens,
cache_read_input_tokens,
cache_creation_input_tokens,
reasoning_tokens: None,
})
}
}
fn collect_content_block_start_text(
data: &str, assistant_text: &mut String, provider_content_blocks: &mut Vec<String>, tx: &Sender<AgentEvent>,
cancel: &CancelToken,
) -> Result<(), String> {
let v: serde_json::Value = serde_json::from_str(data).unwrap_or(serde_json::Value::Null);
let Some(block) = v.get("content_block") else {
return Ok(());
};
match block.get("type").and_then(|t| t.as_str()) {
Some("text") => {
let text = block.get("text").and_then(|t| t.as_str()).unwrap_or("");
if !text.is_empty() {
assistant_text.push_str(text);
if send(tx, AgentEvent::AssistantDelta(text.to_string()), cancel).is_none() {
return Err("cancelled".to_string());
}
}
}
Some("thinking") => {
let thinking = block.get("thinking").and_then(|t| t.as_str()).unwrap_or("");
if !thinking.is_empty() && send(tx, AgentEvent::ReasoningDelta(thinking.to_string()), cancel).is_none() {
return Err("cancelled".to_string());
}
}
Some(other) => {
provider_content_blocks.push(other.to_string());
tracing::info!(content_type = other, "unhandled content block start");
}
None => {}
}
Ok(())
}
fn extract_tool_use_start(data: &str) -> Option<(usize, ToolUseBuilder)> {
let v: serde_json::Value = serde_json::from_str(data).ok()?;
let index = v.get("index").and_then(|i| i.as_u64()).unwrap_or(0) as usize;
let cb = v.get("content_block")?;
let block_type = cb.get("type").and_then(|t| t.as_str())?;
if block_type != "tool_use" {
return None;
}
let name = cb.get("name").and_then(|n| n.as_str())?.to_string();
let id = cb.get("id").and_then(|n| n.as_str()).unwrap_or("").to_string();
let initial_input = cb.get("input").cloned().unwrap_or(serde_json::Value::Null);
Some((
index,
ToolUseBuilder { id, name, initial_input, input_json: String::new() },
))
}
fn send(tx: &Sender<AgentEvent>, event: AgentEvent, cancel: &CancelToken) -> Option<()> {
if cancel.is_cancelled() && !matches!(event, AgentEvent::Cancelled) {
let _ = tx.send(AgentEvent::Cancelled);
return None;
}
if tx.send(event).is_err() {
return None;
}
Some(())
}
fn step() {
thread::sleep(Duration::from_millis(40));
}
#[cfg(test)]
mod tests {
use super::*;
use crate::app::ToolStatus;
use crate::cli::WebSearchMode;
use crate::providers;
use crate::tools::{self, AgentRunConfig, MAX_TOOL_ITERATIONS};
use std::path::{Path, PathBuf};
fn config() -> AgentRunConfig {
AgentRunConfig::new(
PathBuf::from("."),
String::from("fake-agent"),
WebSearchMode::DuckDuckGo,
)
}
struct MetadataErrorProvider {
code: u16,
}
impl StreamingProvider for MetadataErrorProvider {
type Metadata = ();
fn name(&self) -> &'static str {
"metadata-test"
}
fn load_status(&self) -> String {
"provider: loading metadata-test".to_string()
}
fn request_status(&self, _model: &str) -> String {
"provider: requesting metadata-test".to_string()
}
fn from_env_or_dotenv(_root: &Path) -> providers::Result<Self> {
Ok(Self { code: 401 })
}
fn load_metadata(&self) -> providers::Result<Self::Metadata> {
Err(ProviderError::Status { code: self.code, body: "metadata endpoint rejected request".to_string() })
}
fn token_budget(&self, _model: &str, _metadata: Option<&Self::Metadata>) -> u32 {
1
}
fn serialized_request_body(
&self, _model: &str, _messages: &[ProviderMessage], _request: &StreamingRequest<'_>,
) -> providers::Result<Vec<u8>> {
panic!("a rejected metadata request must abort before serializing the prompt")
}
fn send_streaming_request(
&self, _model: &str, _messages: &[ProviderMessage], _request: &StreamingRequest<'_>,
) -> providers::Result<ureq::http::Response<ureq::Body>> {
panic!("a rejected metadata request must abort before sending the prompt")
}
fn stream_format(&self, _model: &str) -> providers::Result<StreamFormat> {
Ok(StreamFormat::AnthropicMessages)
}
fn request_error_message(error: &ProviderError) -> String {
error.failure_message("metadata-test rate limit")
}
fn is_retryable_request_error(error: &ProviderError) -> bool {
error.is_retryable()
}
}
fn dispatch_output(req: &ToolUseRequest, root: &Path) -> tools::ToolOutput {
tools::dispatch_full(req, root).0
}
#[test]
fn provider_retry_policy_and_classification_match_defaults() {
assert_eq!(PROVIDER_RETRY_POLICY.max_retries, 4);
assert_eq!(PROVIDER_RETRY_POLICY.delay_for_attempt(1), Duration::from_millis(2500));
assert_eq!(
PROVIDER_RETRY_POLICY.delay_for_attempt(4),
Duration::from_millis(20_000)
);
assert!(
ProviderAttemptError::Request(providers::ProviderError::Status {
code: 503,
body: "temporarily unavailable".to_string(),
})
.is_retryable::<umans::UmansClient>()
);
assert!(
ProviderAttemptError::Stream("stream read error: connection lost".to_string())
.is_retryable::<umans::UmansClient>()
);
assert!(
!ProviderAttemptError::Request(providers::ProviderError::Status {
code: 401,
body: "unauthorized".to_string()
})
.is_retryable::<umans::UmansClient>()
);
assert!(
!ProviderAttemptError::Stream(
"provider stopped at max_tokens (32768) before producing assistant text".to_string()
)
.is_retryable::<umans::UmansClient>()
);
}
#[test]
fn rejected_metadata_aborts_before_sending_the_prompt() {
let (_steering_tx, steering_rx) = mpsc::channel();
let handle = RunHandle::provider_with_steering(config(), Vec::new(), false, steering_rx);
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
handle.run_provider::<MetadataErrorProvider>(&tx, &cancel);
let events: Vec<AgentEvent> = rx.try_iter().collect();
assert!(events.iter().any(|event| {
matches!(event, AgentEvent::Failed(message) if message == "authentication failed (HTTP 401)")
}));
assert!(!events.iter().any(|event| {
matches!(event, AgentEvent::Status(message) if message.contains("fallback token budget"))
}));
}
#[test]
fn unavailable_metadata_uses_the_fallback_token_budget() {
let provider = MetadataErrorProvider { code: 503 };
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
let metadata = load_provider_metadata(&provider, "metadata-test", &tx, &cancel);
assert!(matches!(metadata, MetadataLoaded::Unavailable));
assert_eq!(
rx.try_recv().expect("fallback status"),
AgentEvent::Status("provider: model metadata unavailable; using fallback token budget".to_string())
);
}
#[test]
fn fake_stream_emits_expected_sequence() {
let handle = RunHandle::fake(config(), String::new());
let rx = handle.spawn();
let mut events = Vec::new();
while let Ok(event) = rx.recv() {
events.push(event);
}
assert_eq!(events.first(), Some(&AgentEvent::Started));
assert_eq!(events.last(), Some(&AgentEvent::Finished));
assert!(events.iter().any(|e| matches!(e, AgentEvent::ReasoningDelta(_))));
assert!(events.iter().any(|e| matches!(e, AgentEvent::AssistantDelta(_))));
assert!(events.iter().any(|e| matches!(e, AgentEvent::ToolStarted { .. })));
assert!(events.iter().any(|e| matches!(e, AgentEvent::ToolFinished { .. })));
}
#[test]
fn fake_stream_with_duckduckgo_search_emits_search_tool_event() {
let mut cfg = config();
cfg.search_mode = WebSearchMode::DuckDuckGo;
let handle = RunHandle::fake(cfg, String::new());
let rx = handle.spawn();
let mut events = Vec::new();
while let Ok(event) = rx.recv() {
events.push(event);
}
let has_search = events
.iter()
.any(|e| matches!(e, AgentEvent::ToolStarted { name, .. } if name == "web_search"));
assert!(has_search, "DuckDuckGo search should emit web_search tool event");
}
#[test]
fn fake_stream_with_none_search_skips_search_and_returns_assistant_text() {
let mut cfg = config();
cfg.search_mode = WebSearchMode::None;
let handle = RunHandle::fake(cfg, String::new());
let rx = handle.spawn();
let mut events = Vec::new();
while let Ok(event) = rx.recv() {
events.push(event);
}
let has_search = events
.iter()
.any(|e| matches!(e, AgentEvent::ToolStarted { name, .. } if name == "web_search"));
assert!(!has_search, "none search should not emit web_search tool event");
assert!(
events.iter().any(|e| matches!(e, AgentEvent::AssistantDelta(_))),
"search-disabled prompt should still return assistant text"
);
assert_eq!(events.last(), Some(&AgentEvent::Finished));
}
#[test]
fn fake_stream_drops_cleanly_when_receiver_dropped() {
let handle = RunHandle::fake(config(), String::new());
let rx = handle.spawn();
drop(rx);
}
#[test]
fn cancel_token_signals_cancellation() {
let token = CancelToken::new();
assert!(!token.is_cancelled());
token.cancel();
assert!(token.is_cancelled());
}
#[test]
fn cancel_terminates_run_without_finishing() {
let handle = RunHandle::fake(config(), String::new());
handle.cancel.cancel();
let rx = handle.spawn();
let mut events = Vec::new();
while let Ok(event) = rx.recv() {
events.push(event);
}
assert!(
!events.contains(&AgentEvent::Finished),
"cancelled run must not finish normally"
);
assert!(
events.contains(&AgentEvent::Cancelled),
"cancelled run must notify the UI before its channel closes"
);
}
#[test]
fn cancellation_notification_is_sent_after_token_is_cancelled() {
let (tx, rx) = mpsc::channel();
let token = CancelToken::new();
token.cancel();
assert_eq!(send(&tx, AgentEvent::Cancelled, &token), Some(()));
assert_eq!(rx.recv().expect("cancellation event"), AgentEvent::Cancelled);
}
#[test]
fn dispatch_tool_find_files_success() {
let req = ToolUseRequest::new(
String::from("find_files"),
serde_json::json!({ "pattern": "mod.rs" }).to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("src/cli"));
assert_eq!(output.status, ToolStatus::Ok);
assert!(output.display.lines.iter().any(|p| p.contains("cli/mod.rs")));
}
#[test]
fn dispatch_tool_read_file_range_success() {
let req = ToolUseRequest::new(
String::from("read_file_range"),
serde_json::json!({
"path": "Cargo.toml",
"start_line": 1,
"end_line": 3
})
.to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Ok);
assert_eq!(output.display.lines.len(), 3);
}
#[test]
fn dispatch_tool_unknown_name_fails() {
let req = ToolUseRequest::new(
String::from("nonexistent_tool"),
String::from("{}"),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Failed);
assert!(output.error.as_ref().is_some_and(|e| e.contains("unknown tool")));
}
#[test]
fn dispatch_tool_malformed_arguments_falls_back_to_defaults() {
let req = ToolUseRequest::new(
String::from("find_files"),
String::from("not valid json"),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("src"));
assert_eq!(output.status, ToolStatus::Ok);
}
#[test]
fn max_tool_iterations_is_reasonable() {
let cap = MAX_TOOL_ITERATIONS;
assert!(cap >= 4 && cap <= 16, "per-turn cap should be 4..=16, got {cap}");
}
#[test]
fn extract_tool_use_start_returns_none_for_text_block() {
let data = r#"{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#;
assert!(extract_tool_use_start(data).is_none());
}
#[test]
fn extract_tool_use_start_returns_builder_for_tool_use_block() {
let data = r#"{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"find_files","input":{"pattern":"cli"}}}"#;
let (_index, block) = extract_tool_use_start(data).expect("should extract");
let req = block.finish().expect("should finish");
assert_eq!(req.name, "find_files");
assert!(req.arguments.contains("cli"));
assert_eq!(req.tool_use_id, "toolu_1");
}
#[test]
fn parse_tool_use_fixture_extracts_request_with_id() {
let sse = include_str!("./providers/fixtures/tool_use_turn.sse");
let chunks = anthropic::parse_sse_chunk(sse);
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
for (event_type, data) in &chunks {
let sse_event = anthropic::parse_sse_event(event_type, data);
if let anthropic::SseEvent::Other(ref t) = sse_event
&& t.starts_with("content_block_start")
&& let Some((_index, block)) = extract_tool_use_start(data)
&& let Some(req) = block.finish()
{
tool_requests.push(req);
}
if let anthropic::SseEvent::TextDelta(ref text) = sse_event {
assistant_text.push_str(text);
}
}
assert_eq!(tool_requests.len(), 1);
let req = &tool_requests[0];
assert_eq!(req.name, "find_files");
assert_eq!(req.tool_use_id, "toolu_01");
assert!(req.arguments.contains("Cargo"));
assert_eq!(assistant_text, "Let me look that up.");
}
#[test]
fn collect_umans_event_reconstructs_streamed_tool_input_json() {
let (tx, _rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut state = AnthropicStreamState::default();
state.collect(
"content_block_start",
r#"{"type":"content_block_start","index":1,"content_block":{"type":"tool_use","id":"toolu_1","name":"find_files","input":{}}}"#,
&tx,
&cancel,
)
.expect("collect event");
state
.collect(
"content_block_delta",
&serde_json::json!({
"type": "content_block_delta",
"index": 1,
"delta": {
"type": "input_json_delta",
"partial_json": "{\"pattern\""
}
})
.to_string(),
&tx,
&cancel,
)
.expect("collect event");
state
.collect(
"content_block_delta",
&serde_json::json!({
"type": "content_block_delta",
"index": 1,
"delta": {
"type": "input_json_delta",
"partial_json": ":\"Cargo\"}"
}
})
.to_string(),
&tx,
&cancel,
)
.expect("collect event");
state
.collect(
"content_block_stop",
r#"{"type":"content_block_stop","index":1}"#,
&tx,
&cancel,
)
.expect("collect event");
assert_eq!(state.tool_requests.len(), 1);
assert_eq!(state.tool_requests[0].name, "find_files");
assert_eq!(state.tool_requests[0].tool_use_id, "toolu_1");
assert_eq!(state.tool_requests[0].arguments, r#"{"pattern":"Cargo"}"#);
}
#[test]
fn collect_umans_event_tracks_provider_side_content_blocks() {
let (tx, _rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut state = AnthropicStreamState::default();
state.collect(
"content_block_start",
r#"{"type":"content_block_start","index":0,"content_block":{"type":"server_tool_use","id":"srv_1","name":"web_search","input":{}}}"#,
&tx,
&cancel,
)
.expect("collect event");
state.collect(
"content_block_start",
r#"{"type":"content_block_start","index":1,"content_block":{"type":"web_search_tool_result","content":[]}}"#,
&tx,
&cancel,
)
.expect("collect event");
assert_eq!(
state.provider_content_blocks,
vec!["server_tool_use".to_string(), "web_search_tool_result".to_string()]
);
assert!(state.assistant_text.is_empty());
assert!(state.tool_requests.is_empty());
}
#[test]
fn collect_openai_chat_event_maps_text_reasoning_and_usage() {
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_openai_chat_event(
openai::ChatSseEvent::TextDelta("hello".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("text delta");
collect_openai_chat_event(
openai::ChatSseEvent::ReasoningDelta("thinking".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("reasoning delta");
collect_openai_chat_event(
openai::ChatSseEvent::Usage { input_tokens: 2, output_tokens: 3 },
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("usage");
assert_eq!(assistant_text, "hello");
let events: Vec<AgentEvent> = rx.try_iter().collect();
assert!(events.contains(&AgentEvent::AssistantDelta("hello".to_string())));
assert!(events.contains(&AgentEvent::ReasoningDelta("thinking".to_string())));
assert!(!events.iter().any(|event| matches!(event, AgentEvent::Usage { .. })));
}
#[test]
fn collect_openai_chat_event_finishes_tool_calls_on_finish_reason() {
let (tx, _rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_openai_chat_event(
openai::ChatSseEvent::ToolCallStart { index: 0, id: "call_1".to_string(), name: "find_files".to_string() },
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("tool start");
collect_openai_chat_event(
openai::ChatSseEvent::ToolCallArgumentsDelta { index: 0, arguments: r#"{"pattern":"Cargo"}"#.to_string() },
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("tool args");
collect_openai_chat_event(
openai::ChatSseEvent::FinishReason("tool_calls".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("finish");
assert_eq!(stop_reason.as_deref(), Some("tool_calls"));
assert_eq!(tool_requests.len(), 1);
assert_eq!(tool_requests[0].name, "find_files");
assert_eq!(tool_requests[0].tool_use_id, "call_1");
assert_eq!(tool_requests[0].arguments, r#"{"pattern":"Cargo"}"#);
}
#[test]
fn collect_openai_chat_event_handles_status_and_failures() {
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_openai_chat_event(
openai::ChatSseEvent::ResponseStatus("queued".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("queued status");
assert!(matches!(
rx.try_recv(),
Ok(AgentEvent::Status(status)) if status == "provider: status queued"
));
let failed = collect_openai_chat_event(
openai::ChatSseEvent::ResponseStatus("failed".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("failed status should fail");
assert!(failed.contains("provider stream status: failed"));
let backend = collect_openai_chat_event(
openai::ChatSseEvent::Error("backend failed".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("backend error should fail");
assert!(backend.contains("provider error: backend failed"));
let malformed = collect_openai_chat_event(
openai::ChatSseEvent::Malformed("{bad".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("malformed payload should fail");
assert!(malformed.contains("malformed provider stream payload"));
assert!(!ProviderAttemptError::Stream(malformed).is_retryable::<umans::UmansClient>());
}
#[test]
fn collect_chatgpt_codex_event_maps_text_reasoning_and_usage() {
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::TextDelta("hello".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("text delta");
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ReasoningDelta("thinking".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("reasoning delta");
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::Usage { input_tokens: 3, output_tokens: 5 },
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("usage");
assert_eq!(assistant_text, "hello");
let events: Vec<AgentEvent> = rx.try_iter().collect();
assert!(events.contains(&AgentEvent::AssistantDelta("hello".to_string())));
assert!(events.contains(&AgentEvent::ReasoningDelta("thinking".to_string())));
assert!(!events.iter().any(|event| matches!(event, AgentEvent::Usage { .. })));
}
#[test]
fn collect_chatgpt_codex_event_finishes_tool_calls_on_done() {
let (tx, _rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ToolCallStart {
id: "fc_1".to_string(),
call_id: "call_1".to_string(),
name: "find_files".to_string(),
},
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("tool start");
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ToolCallArgumentsDelta {
id: "fc_1".to_string(),
arguments: r#"{"pattern":"Car"#.to_string(),
},
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("tool args");
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ToolCallDone {
id: "fc_1".to_string(),
call_id: Some("call_1".to_string()),
name: "find_files".to_string(),
arguments: r#"{"pattern":"Cargo"}"#.to_string(),
},
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("tool done");
assert!(tool_blocks.is_empty());
assert_eq!(tool_requests.len(), 1);
assert_eq!(tool_requests[0].name, "find_files");
assert_eq!(tool_requests[0].tool_use_id, "call_1");
assert_eq!(tool_requests[0].arguments, r#"{"pattern":"Cargo"}"#);
}
#[test]
fn collect_chatgpt_codex_event_handles_statuses_and_failures() {
let (tx, rx) = mpsc::channel();
let cancel = CancelToken::new();
let mut tool_blocks = HashMap::new();
let mut tool_requests = Vec::new();
let mut assistant_text = String::new();
let mut stop_reason = None;
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ResponseStatus("queued".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("queued status");
collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ResponseStatus("completed".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect("completed status");
assert_eq!(stop_reason.as_deref(), Some("completed"));
let events: Vec<AgentEvent> = rx.try_iter().collect();
assert!(events.contains(&AgentEvent::Status("provider: status queued".to_string())));
assert!(events.contains(&AgentEvent::Status("provider: status completed".to_string())));
let failed = collect_chatgpt_codex_event(
codex::ResponsesSseEvent::ResponseStatus("incomplete".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("incomplete status should fail");
assert!(failed.contains("provider stream status: incomplete"));
let backend = collect_chatgpt_codex_event(
codex::ResponsesSseEvent::Error("backend failed".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("backend error should fail");
assert!(backend.contains("provider error: backend failed"));
let malformed = collect_chatgpt_codex_event(
codex::ResponsesSseEvent::Malformed("{bad".to_string()),
&mut tool_blocks,
&mut tool_requests,
&mut assistant_text,
&mut stop_reason,
&tx,
&cancel,
)
.expect_err("malformed payload should fail");
assert!(malformed.contains("malformed provider stream payload"));
assert!(!ProviderAttemptError::Stream(malformed).is_retryable::<umans::UmansClient>());
}
#[test]
fn append_steering_messages_adds_user_messages() {
let (tx, rx) = mpsc::channel();
tx.send("look at tests first".to_string()).expect("send steering");
drop(tx);
let handle = RunHandle::provider_with_steering(config(), Vec::new(), false, rx);
let mut messages = Vec::new();
assert!(append_steering_messages(&mut messages, &handle));
assert_eq!(messages.len(), 1);
assert_eq!(messages[0].role, "user");
assert!(messages[0].as_text().contains("[steering]"));
assert!(messages[0].as_text().contains("look at tests first"));
}
fn handle_with_permission(decision: ToolPermissionDecision) -> RunHandle {
let (_tx, rx) = mpsc::channel();
RunHandle::provider_with_steering(config(), Vec::new(), false, rx)
.with_permission_hook(ToolPermissionHook::new(move |_request, _config, _cancel| decision))
}
fn approve_request(name: &str, decision: ToolPermissionDecision) -> ToolPermissionDecision {
let request = ToolUseRequest::new(name.to_string(), "{}".to_string(), "tool-1".to_string());
let handle = handle_with_permission(decision);
approve_tool_request(&request, &handle, &CancelToken::new())
}
#[test]
fn permission_hook_allows_file_write_tool() {
assert_eq!(
approve_request("create_file", ToolPermissionDecision::Allow),
ToolPermissionDecision::Allow
);
}
#[test]
fn permission_hook_rejects_file_write_tool() {
assert_eq!(
approve_request("replace_range", ToolPermissionDecision::Reject),
ToolPermissionDecision::Reject
);
}
#[test]
fn permission_hook_allows_shell_tool() {
assert_eq!(
approve_request("run_shell", ToolPermissionDecision::Allow),
ToolPermissionDecision::Allow
);
}
#[test]
fn permission_hook_rejects_shell_tool() {
assert_eq!(
approve_request("run_shell", ToolPermissionDecision::Reject),
ToolPermissionDecision::Reject
);
}
#[test]
fn permission_hook_cancels_sensitive_tool() {
assert_eq!(
approve_request("write_patch", ToolPermissionDecision::Cancelled),
ToolPermissionDecision::Cancelled
);
}
#[test]
fn read_only_tool_bypasses_permission_hook() {
assert_eq!(
approve_request("read_file_range", ToolPermissionDecision::Reject),
ToolPermissionDecision::Allow
);
}
#[test]
fn prompt_expects_workspace_write_for_file_edit_request() {
assert!(prompt_expects_workspace_write(
"Looking at completed work in TODO.md, can you summarize them like the completed sections?"
));
assert!(prompt_expects_workspace_write("update README.md with install notes"));
}
#[test]
fn prompt_expects_workspace_write_ignores_plain_file_questions() {
assert!(!prompt_expects_workspace_write("what does TODO.md contain?"));
assert!(!prompt_expects_workspace_write("summarize the project architecture"));
}
#[test]
fn provider_kind_routes_open_code_prefixes_separately() {
assert_eq!(
ProviderKind::for_model("opencode/big-pickle"),
ProviderKind::OpenCodeZen
);
assert_eq!(
ProviderKind::for_model("opencode-go/kimi-k2.7-code"),
ProviderKind::OpenCodeGo
);
assert_eq!(
ProviderKind::for_model("chatgpt-codex/gpt-5.5"),
ProviderKind::ChatGptCodex
);
assert_eq!(ProviderKind::for_model("big-pickle"), ProviderKind::Umans);
}
#[test]
fn dispatch_read_url_rejects_private_network() {
let req = ToolUseRequest::new(
String::from("read_url"),
serde_json::json!({ "url": "http://127.0.0.1/secret" }).to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Failed);
assert!(output.error.as_ref().is_some_and(|e| e.contains("private network")));
}
#[test]
fn dispatch_read_url_rejects_non_public_scheme() {
let req = ToolUseRequest::new(
String::from("read_url"),
serde_json::json!({ "url": "file:///etc/passwd" }).to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Failed);
assert!(output.error.as_ref().is_some_and(|e| e.contains("unsupported")));
}
#[test]
fn tool_definitions_include_web_search_and_read_url() {
let defs = tools::tool_definitions();
let names = defs.iter().map(|d| d.name.as_ref()).collect::<Vec<&str>>();
assert!(names.contains(&"web_search"));
assert!(names.contains(&"read_url"));
assert!(names.contains(&"run_shell"), "tool catalog should include run_shell");
}
#[test]
fn dispatch_run_shell_success() {
let req = ToolUseRequest::new(
String::from("run_shell"),
serde_json::json!({ "program": "echo", "args": ["hello"] }).to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Ok);
assert_eq!(output.name, "run_shell");
assert!(output.display.lines.iter().any(|l| l.contains("hello")));
}
#[test]
fn dispatch_run_shell_failure() {
let req = ToolUseRequest {
name: String::from("run_shell"),
arguments: serde_json::json!({ "program": "sh", "args": ["-c", "exit 1"] }).to_string(),
tool_use_id: String::from("toolu_test"),
};
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Failed);
assert!(output.error.as_ref().is_some_and(|e| e.contains("exit 1")));
}
#[test]
fn dispatch_run_shell_missing_program_fails() {
let req = ToolUseRequest::new(
String::from("run_shell"),
serde_json::json!({ "args": ["test"] }).to_string(),
String::from("toolu_test"),
);
let output = dispatch_output(&req, Path::new("."));
assert_eq!(output.status, ToolStatus::Failed);
assert!(output.error.as_ref().is_some_and(|e| e.contains("missing")));
}
#[test]
fn agent_dispatch_cancellation_stops_run_shell() {
let dir = tempfile::tempdir().expect("temp dir");
let handle = RunHandle::fake(
AgentRunConfig::new(dir.path().to_path_buf(), "fake-agent".to_string(), WebSearchMode::None),
String::new(),
);
let request = ToolUseRequest::new(
"run_shell".to_string(),
serde_json::json!({ "argv": ["sh", "-c", "exec sleep 30"] }).to_string(),
"call_1".to_string(),
);
let cancel = handle.cancel.clone();
let canceller = cancel.clone();
let stopper = std::thread::spawn(move || {
std::thread::sleep(std::time::Duration::from_millis(100));
canceller.cancel();
});
let started = std::time::Instant::now();
let (output, _, process) = dispatch_tool_request(&request, &handle, &cancel);
stopper.join().expect("cancellation thread");
let process = process.expect("shell process result");
assert_eq!(output.status, ToolStatus::Failed);
assert_eq!(process.status, crate::tools::shell::ProcessStatus::Cancelled);
assert!(started.elapsed() < std::time::Duration::from_secs(5));
}
#[test]
fn tool_display_output_includes_failure_detail_without_changing_success_output() {
let success = ToolOutput::ok("find_files", vec!["src/lib.rs".to_string()]);
assert_eq!(success.display_lines(), vec!["src/lib.rs"]);
let failure = ToolOutput::failed(
"run_shell",
"missing command: provide non-empty 'argv', 'command', or 'program'",
);
assert_eq!(
failure.display_lines(),
vec!["error: missing command: provide non-empty 'argv', 'command', or 'program'"]
);
}
}