use async_trait::async_trait;
use eventsource_stream::Eventsource;
use futures::stream::{BoxStream, StreamExt};
use serde_json::{json, Value};
use crate::event::HarnessUsage;
use crate::model_catalog::{ReasoningMode, ResolvedModelConfig, WireProtocol};
use crate::tools::{ToolInvocation, ToolSpec};
#[derive(Debug, Clone, PartialEq)]
pub enum ModelChunk {
TextDelta {
msg_id: String,
delta: String,
},
ThinkingDelta {
thinking_id: String,
delta: String,
signature: Option<String>,
},
ToolCallStart {
id: String,
name: String,
},
ToolCallInputDelta {
id: String,
delta: String,
},
ToolCallEnd {
id: String,
input: Option<Value>,
},
Done {
stop_reason: String,
usage: Option<HarnessUsage>,
},
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct AssistantThinking {
pub text: String,
pub signature: Option<String>,
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct ImageSource {
pub media_type: String,
pub data: ImageData,
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum ImageData {
Base64(String),
Url(String),
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum UserAttachment {
Image(ImageSource),
}
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub enum ChatMessage {
User {
content: String,
attachments: Vec<UserAttachment>,
},
Assistant {
text: Option<String>,
tool_calls: Vec<ToolInvocation>,
thinking: Option<AssistantThinking>,
#[serde(default, skip_serializing_if = "Option::is_none")]
usage: Option<crate::event::HarnessUsage>,
},
Tool {
tool_call_id: String,
content: String,
is_error: bool,
attachments: Vec<UserAttachment>,
},
}
#[derive(Debug, Clone, PartialEq)]
pub struct ModelTurnInput {
pub system_prompt: Option<String>,
pub messages: Vec<ChatMessage>,
pub tools: Vec<ToolSpec>,
pub hosted_tools: Vec<HostedTool>,
pub tool_choice: ToolChoice,
pub parallel_tool_calls: Option<bool>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum HostedCapability {
WebSearch,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum CapabilitySupport {
Supported,
Unsupported,
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, serde::Serialize, serde::Deserialize)]
pub enum HostedTool {
WebSearch,
}
#[derive(Debug, Clone, PartialEq, Default)]
pub enum ToolChoice {
#[default]
Auto,
None,
Required,
Tool(String),
}
impl ToolChoice {
pub fn parse(s: &str) -> Self {
let trimmed = s.trim();
if let Some(name) = trimmed.strip_prefix("tool:") {
return Self::Tool(name.trim().to_string());
}
match trimmed.to_ascii_lowercase().as_str() {
"" | "auto" => Self::Auto,
"none" => Self::None,
"required" | "any" => Self::Required,
_ => Self::Auto,
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub enum ModelResponse {
Message {
text: String,
stop_reason: String,
usage: Option<HarnessUsage>,
},
ToolCall {
preface: Option<String>,
invocation: ToolInvocation,
usage: Option<HarnessUsage>,
},
}
impl ModelResponse {
pub fn usage(&self) -> Option<&HarnessUsage> {
match self {
ModelResponse::Message { usage, .. } | ModelResponse::ToolCall { usage, .. } => {
usage.as_ref()
}
}
}
}
#[derive(Debug, thiserror::Error)]
pub enum ModelClientError {
#[error("rate limit: {0}")]
RateLimit(String),
#[error("auth: {0}")]
Auth(String),
#[error("context overflow: {0}")]
ContextOverflow(String),
#[error("bad request: {0}")]
BadRequest(String),
#[error("server error: {0}")]
ServerError(String),
#[error("network: {0}")]
Network(String),
#[error("model error: {0}")]
Other(String),
}
impl ModelClientError {
pub fn retryable(&self) -> bool {
matches!(
self,
Self::RateLimit(_) | Self::Network(_) | Self::ServerError(_)
)
}
}
#[async_trait]
pub trait ModelClient: Send + Sync {
fn hosted_capability(&self, capability: HostedCapability) -> CapabilitySupport;
async fn stream(
&self,
input: ModelTurnInput,
) -> Result<BoxStream<'static, Result<ModelChunk, ModelClientError>>, ModelClientError>;
async fn next(&self, input: ModelTurnInput) -> Result<ModelResponse, ModelClientError> {
let stream = self.stream(input).await?;
collect_model_response(stream).await
}
}
pub async fn collect_model_response(
mut stream: BoxStream<'static, Result<ModelChunk, ModelClientError>>,
) -> Result<ModelResponse, ModelClientError> {
let mut text_buf = String::new();
let mut text_msg_id: Option<String> = None;
let mut tool_states: Vec<ToolStreamState> = Vec::new();
let mut stop_reason: Option<String> = None;
let mut usage: Option<HarnessUsage> = None;
while let Some(item) = stream.next().await {
match item? {
ModelChunk::TextDelta { msg_id, delta } => {
if text_msg_id.as_deref() != Some(&msg_id) {
text_msg_id = Some(msg_id);
text_buf.clear();
}
text_buf.push_str(&delta);
}
ModelChunk::ThinkingDelta { .. } => {
}
ModelChunk::ToolCallStart { id, name } => {
tool_states.push(ToolStreamState {
id,
name,
args_buf: String::new(),
early_input: None,
});
}
ModelChunk::ToolCallInputDelta { id, delta } => {
if let Some(state) = tool_states.iter_mut().find(|s| s.id == id) {
state.args_buf.push_str(&delta);
}
}
ModelChunk::ToolCallEnd { id, input } => {
if let Some(state) = tool_states.iter_mut().find(|s| s.id == id) {
state.early_input = input;
}
}
ModelChunk::Done {
stop_reason: sr,
usage: u,
} => {
stop_reason = Some(sr);
usage = u;
}
}
}
if let Some(state) = tool_states.into_iter().next() {
let parsed_input = match state.early_input {
Some(v) => v,
None => serde_json::from_str(state.args_buf.as_str().trim()).map_err(|e| {
ModelClientError::Other(format!(
"decode tool arguments for {id}: {e}",
id = state.id
))
})?,
};
let raw_emitted_args = raw_args_for_input(&state.args_buf, &parsed_input);
return Ok(ModelResponse::ToolCall {
preface: (!text_buf.is_empty()).then(|| text_buf.clone()),
invocation: ToolInvocation {
id: state.id,
name: state.name,
input: parsed_input,
raw_emitted_args,
},
usage,
});
}
Ok(ModelResponse::Message {
text: text_buf,
stop_reason: stop_reason.unwrap_or_else(|| "end_turn".into()),
usage,
})
}
struct ToolStreamState {
id: String,
name: String,
args_buf: String,
early_input: Option<Value>,
}
fn raw_args_for_input(raw: &str, input: &Value) -> Option<String> {
let trimmed = raw.trim();
if trimmed.is_empty() {
return None;
}
match serde_json::from_str::<Value>(trimmed) {
Ok(parsed) if parsed == *input => Some(trimmed.to_string()),
_ => None,
}
}
fn tool_invocation_args_for_wire(tc: &ToolInvocation) -> String {
tc.raw_emitted_args
.as_deref()
.and_then(|raw| raw_args_for_input(raw, &tc.input))
.unwrap_or_else(|| tc.input.to_string())
}
#[derive(Debug, Clone)]
pub struct OpenAiCompatibleConfig {
pub base_url: String,
pub api_key: String,
pub model: ResolvedModelConfig,
}
#[derive(Debug, Clone)]
pub struct OpenAiCompatibleModelClient {
http: reqwest::Client,
config: OpenAiCompatibleConfig,
}
impl OpenAiCompatibleModelClient {
pub fn new(config: OpenAiCompatibleConfig) -> Self {
assert_eq!(
config.model.wire_protocol,
WireProtocol::OpenAiCompatible,
"resolved model protocol must match OpenAiCompatibleModelClient"
);
let http = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(15))
.build()
.unwrap_or_else(|_| reqwest::Client::new());
Self { http, config }
}
fn endpoint(&self) -> String {
let base = self.config.base_url.trim_end_matches('/');
if base.ends_with("/chat/completions") {
base.to_string()
} else {
format!("{base}/chat/completions")
}
}
fn request_body(&self, input: &ModelTurnInput) -> Value {
let mut messages = Vec::with_capacity(input.messages.len() + 1);
if let Some(sys) = input.system_prompt.as_deref().filter(|s| !s.is_empty()) {
messages.push(json!({ "role": "system", "content": sys }));
}
for msg in &input.messages {
messages.push(chat_message_to_wire(msg));
}
let mut body = json!({
"model": self.config.model.model,
"messages": messages,
});
let send_tools = !input.tools.is_empty() && !matches!(input.tool_choice, ToolChoice::None);
if send_tools {
body["tools"] = json!(input
.tools
.iter()
.map(tool_spec_to_openai_function)
.collect::<Vec<_>>());
body["tool_choice"] = openai_tool_choice_value(&input.tool_choice);
if let Some(parallel) = input.parallel_tool_calls {
body["parallel_tool_calls"] = json!(parallel);
}
}
if let Some(temperature) = self.config.model.temperature {
body["temperature"] = json!(temperature);
}
body["max_tokens"] = json!(self.config.model.max_output_tokens);
apply_openai_compatible_reasoning(&mut body, &self.config.model);
body
}
}
fn apply_openai_compatible_reasoning(body: &mut Value, model: &ResolvedModelConfig) {
if matches!(model.reasoning.mode, ReasoningMode::Default) {
return;
}
if let Some(effort) = model.reasoning.effort.as_deref() {
body["reasoning_effort"] = json!(effort);
return;
}
let mut thinking = json!({
"type": if matches!(model.reasoning.mode, ReasoningMode::Enabled) {
"enabled"
} else {
"disabled"
}
});
if let Some(tokens) = model.reasoning.budget_tokens {
thinking["budget_tokens"] = json!(tokens);
}
body["thinking"] = thinking;
}
fn openai_tool_choice_value(c: &ToolChoice) -> Value {
match c {
ToolChoice::Auto => json!("auto"),
ToolChoice::None => json!("none"),
ToolChoice::Required => json!("required"),
ToolChoice::Tool(name) => json!({
"type": "function",
"function": {"name": name},
}),
}
}
fn parse_openai_usage(usage: Option<&Value>) -> Option<HarnessUsage> {
let u = usage?;
let input = u.get("prompt_tokens").and_then(|v| v.as_u64()).unwrap_or(0);
let output = u
.get("completion_tokens")
.and_then(|v| v.as_u64())
.unwrap_or(0);
let cache_read = u
.get("prompt_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(|v| v.as_u64())
.unwrap_or(0);
if input == 0 && output == 0 && cache_read == 0 {
return None;
}
Some(HarnessUsage {
input_tokens: input,
output_tokens: output,
cache_read_input_tokens: cache_read,
cache_creation_input_tokens: 0,
compaction_input_tokens: 0,
compaction_output_tokens: 0,
})
}
fn image_to_openai_part(src: &ImageSource) -> Value {
let url = match &src.data {
ImageData::Base64(b64) => {
format!("data:{};base64,{}", src.media_type, b64)
}
ImageData::Url(u) => u.clone(),
};
json!({
"type": "image_url",
"image_url": { "url": url },
})
}
fn tool_spec_to_openai_function(spec: &ToolSpec) -> Value {
json!({
"type": "function",
"function": {
"name": spec.name,
"description": spec.description,
"parameters": spec.input_schema,
}
})
}
const MAX_TOOL_RESULT_REPLAY_TOKENS: u64 = 2_000;
const MAX_TOOL_RESULT_REPLAY_BYTES: usize = 12 * 1024;
const COMPACTED_TOOL_RESULT_KEEP_CHARS: usize = 3_000;
fn compact_tool_result_for_replay(content: &str) -> std::borrow::Cow<'_, str> {
let estimated_tokens = crate::compaction::estimate_tokens(content);
if estimated_tokens <= MAX_TOOL_RESULT_REPLAY_TOKENS
&& content.len() <= MAX_TOOL_RESULT_REPLAY_BYTES
{
return std::borrow::Cow::Borrowed(content);
}
let chars: Vec<char> = content.chars().collect();
if chars.len() <= COMPACTED_TOOL_RESULT_KEEP_CHARS {
return std::borrow::Cow::Borrowed(content);
}
let head_len = COMPACTED_TOOL_RESULT_KEEP_CHARS / 2;
let tail_len = COMPACTED_TOOL_RESULT_KEEP_CHARS - head_len;
let head: String = chars[..head_len].iter().collect();
let tail: String = chars[chars.len() - tail_len..].iter().collect();
let omitted = chars.len() - COMPACTED_TOOL_RESULT_KEEP_CHARS;
std::borrow::Cow::Owned(format!(
"[tool result compacted for model replay]\n\
original_estimated_tokens={estimated_tokens} original_chars={} \
retained_head_chars={head_len} retained_tail_chars={tail_len}\n\
The full raw tool result remains in session history; this replay is abbreviated.\n\n\
--- head ---\n{head}\n\n\
--- omitted ---\n[... omitted {omitted} chars from tool result replay ...]\n\n\
--- tail ---\n{tail}",
chars.len(),
))
}
fn chat_message_to_wire(msg: &ChatMessage) -> Value {
match msg {
ChatMessage::User {
content,
attachments,
} => {
if attachments.is_empty() {
json!({ "role": "user", "content": content })
} else {
let mut parts: Vec<Value> = Vec::with_capacity(attachments.len() + 1);
if !content.is_empty() {
parts.push(json!({ "type": "text", "text": content }));
}
for att in attachments {
match att {
UserAttachment::Image(src) => {
parts.push(image_to_openai_part(src));
}
}
}
json!({ "role": "user", "content": parts })
}
}
ChatMessage::Assistant {
text,
tool_calls,
thinking: _,
usage: _,
} => {
let mut obj = json!({ "role": "assistant" });
if let Some(t) = text.as_deref().filter(|s| !s.is_empty()) {
obj["content"] = json!(t);
} else {
obj["content"] = Value::Null;
}
if !tool_calls.is_empty() {
let calls: Vec<Value> = tool_calls
.iter()
.map(|tc| {
json!({
"id": tc.id,
"type": "function",
"function": {
"name": tc.name,
"arguments": tool_invocation_args_for_wire(tc),
},
})
})
.collect();
obj["tool_calls"] = json!(calls);
}
obj
}
ChatMessage::Tool {
tool_call_id,
content,
attachments,
is_error: _,
} => {
let mut content_str = compact_tool_result_for_replay(content).into_owned();
for att in attachments {
let UserAttachment::Image(src) = att;
content_str.push_str(&format!(
"\n[image attached: {} (not visible via OpenAI tool role)]",
src.media_type
));
}
json!({
"role": "tool",
"tool_call_id": tool_call_id,
"content": content_str,
})
}
}
}
#[async_trait]
impl ModelClient for OpenAiCompatibleModelClient {
fn hosted_capability(&self, _capability: HostedCapability) -> CapabilitySupport {
CapabilitySupport::Unsupported
}
async fn stream(
&self,
input: ModelTurnInput,
) -> Result<BoxStream<'static, Result<ModelChunk, ModelClientError>>, ModelClientError> {
if !input.hosted_tools.is_empty() {
return Err(ModelClientError::Other(
"hosted tools are not supported by OpenAiCompatibleModelClient; \
OpenAI web_search requires a Responses API client"
.into(),
));
}
let mut body = self.request_body(&input);
body["stream"] = json!(true);
body["stream_options"] = json!({ "include_usage": true });
let resp = match self
.http
.post(self.endpoint())
.bearer_auth(&self.config.api_key)
.json(&body)
.send()
.await
{
Ok(r) => r,
Err(e) => return Err(classify_reqwest_error(&e, e.to_string())),
};
let status = resp.status();
if !status.is_success() {
let body_text = resp.text().await.unwrap_or_default();
return Err(classify_openai_http_error(status, &body_text));
}
let event_stream = resp.bytes_stream().eventsource();
let (tx, rx) = tokio::sync::mpsc::channel::<Result<ModelChunk, ModelClientError>>(8);
tokio::spawn(async move {
let mut state = OpenAiStreamState::default();
futures::pin_mut!(event_stream);
while let Some(ev) = event_stream.next().await {
let chunks = match ev {
Ok(event) => match state.feed_data(&event.data) {
Ok(c) => c,
Err(e) => {
let _ = tx.send(Err(e)).await;
return;
}
},
Err(e) => {
let _ = tx
.send(Err(ModelClientError::Network(format!(
"SSE transport error: {e}"
))))
.await;
return;
}
};
for c in chunks {
if tx.send(Ok(c)).await.is_err() {
return;
}
}
}
if state.ended_cleanly() {
if let Some(final_chunk) = state.finalize() {
let _ = tx.send(Ok(final_chunk)).await;
}
} else {
let _ = tx
.send(Err(ModelClientError::Network(
"model stream closed before completion (no finish_reason or [DONE]) \
— connection dropped or upstream truncated the response"
.into(),
)))
.await;
}
});
Ok(tokio_stream::wrappers::ReceiverStream::new(rx).boxed())
}
}
#[derive(Debug, Default)]
struct OpenAiStreamState {
msg_id: Option<String>,
tool_call_by_index: std::collections::HashMap<u64, String>,
finish_reason: Option<String>,
pending_usage: Option<HarnessUsage>,
done_emitted: bool,
}
impl OpenAiStreamState {
fn feed_data(&mut self, data: &str) -> Result<Vec<ModelChunk>, ModelClientError> {
if data.trim() == "[DONE]" {
if let Some(done) = self.emit_done() {
return Ok(vec![done]);
}
return Ok(vec![]);
}
let value: Value = serde_json::from_str(data)
.map_err(|e| ModelClientError::Other(format!("SSE data not JSON: {e}; raw={data}")))?;
let mut out: Vec<ModelChunk> = Vec::new();
if let Some(usage) = parse_openai_usage(value.get("usage")) {
self.pending_usage = Some(usage);
}
if let Some(id) = value.get("id").and_then(|v| v.as_str()) {
if self.msg_id.is_none() && !id.is_empty() {
self.msg_id = Some(id.to_string());
}
}
let Some(choices) = value.get("choices").and_then(|v| v.as_array()) else {
return Ok(out);
};
let Some(choice) = choices.first() else {
return Ok(out);
};
let Some(delta) = choice.get("delta") else {
if let Some(reason) = choice.get("finish_reason").and_then(|v| v.as_str()) {
self.finish_reason = Some(reason.to_string());
}
return Ok(out);
};
if let Some(text) = delta.get("content").and_then(|v| v.as_str()) {
if !text.is_empty() {
let msg_id = self
.msg_id
.clone()
.unwrap_or_else(|| "msg_native_default".to_string());
out.push(ModelChunk::TextDelta {
msg_id,
delta: text.to_string(),
});
}
}
if let Some(tcs) = delta.get("tool_calls").and_then(|v| v.as_array()) {
for tc in tcs {
let index = tc.get("index").and_then(|v| v.as_u64()).unwrap_or(0);
if let Some(id) = tc.get("id").and_then(|v| v.as_str()) {
if !id.is_empty() {
self.tool_call_by_index.insert(index, id.to_string());
let name = tc
.get("function")
.and_then(|f| f.get("name"))
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
out.push(ModelChunk::ToolCallStart {
id: id.to_string(),
name,
});
}
}
if let Some(args) = tc
.get("function")
.and_then(|f| f.get("arguments"))
.and_then(|v| v.as_str())
{
if let Some(id) = self.tool_call_by_index.get(&index).cloned() {
if !args.is_empty() {
out.push(ModelChunk::ToolCallInputDelta {
id,
delta: args.to_string(),
});
}
}
}
}
}
if let Some(reason) = choice.get("finish_reason").and_then(|v| v.as_str()) {
self.finish_reason = Some(reason.to_string());
if reason == "tool_calls" {
for (_idx, id) in self.tool_call_by_index.iter() {
out.push(ModelChunk::ToolCallEnd {
id: id.clone(),
input: None,
});
}
}
}
Ok(out)
}
fn finalize(&mut self) -> Option<ModelChunk> {
self.emit_done()
}
fn ended_cleanly(&self) -> bool {
self.done_emitted || self.finish_reason.is_some()
}
fn emit_done(&mut self) -> Option<ModelChunk> {
if self.done_emitted {
return None;
}
self.done_emitted = true;
let stop_reason = map_openai_finish_reason(self.finish_reason.as_deref());
Some(ModelChunk::Done {
stop_reason,
usage: self.pending_usage.take(),
})
}
}
fn map_openai_finish_reason(reason: Option<&str>) -> String {
match reason {
Some("stop") => "end_turn".into(),
Some("length") => "max_tokens".into(),
Some("tool_calls") => "end_turn".into(),
Some("content_filter") => "refusal".into(),
Some(other) if !other.is_empty() => other.to_string(),
_ => "end_turn".into(),
}
}
fn classify_openai_http_error(status: reqwest::StatusCode, body: &str) -> ModelClientError {
use reqwest::StatusCode;
let snippet = body.chars().take(512).collect::<String>();
if status == StatusCode::TOO_MANY_REQUESTS {
return ModelClientError::RateLimit(format!("HTTP {status}: {snippet}"));
}
if status.is_server_error() {
return ModelClientError::ServerError(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN {
return ModelClientError::Auth(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::BAD_REQUEST && looks_like_context_overflow(body) {
return ModelClientError::ContextOverflow(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::BAD_REQUEST {
return ModelClientError::BadRequest(format!("HTTP {status}: {snippet}"));
}
ModelClientError::Other(format!("HTTP {status}: {snippet}"))
}
fn looks_like_context_overflow(body: &str) -> bool {
let lower = body.to_lowercase();
lower.contains("context length")
|| lower.contains("maximum context")
|| lower.contains("context_length_exceeded")
|| lower.contains("too many tokens")
|| lower.contains("exceeds the model")
}
fn classify_reqwest_error(err: &reqwest::Error, msg: String) -> ModelClientError {
if err.is_connect() || err.is_timeout() || err.is_request() || err.is_body() {
ModelClientError::Network(msg)
} else {
ModelClientError::Other(msg)
}
}
#[derive(Debug, Default, Clone)]
pub struct ScriptedModelClient;
#[async_trait]
impl ModelClient for ScriptedModelClient {
fn hosted_capability(&self, _capability: HostedCapability) -> CapabilitySupport {
CapabilitySupport::Unsupported
}
async fn stream(
&self,
input: ModelTurnInput,
) -> Result<BoxStream<'static, Result<ModelChunk, ModelClientError>>, ModelClientError> {
let chunks = scripted_chunks_for(&input);
let stream = futures::stream::iter(chunks.into_iter().map(Ok));
Ok(stream.boxed())
}
}
fn scripted_chunks_for(input: &ModelTurnInput) -> Vec<ModelChunk> {
let last_tool = input.messages.iter().rev().find_map(|m| match m {
ChatMessage::Tool {
tool_call_id,
content,
is_error,
..
} => Some((tool_call_id.clone(), content.clone(), *is_error)),
_ => None,
});
if let Some((id, content, is_error)) = last_tool {
let summary = if is_error {
format!("tool {id} failed: {content}")
} else {
format!("tool {id} completed: {content}")
};
return vec![
ModelChunk::TextDelta {
msg_id: "scripted_msg".into(),
delta: summary,
},
ModelChunk::Done {
stop_reason: "end_turn".into(),
usage: None,
},
];
}
let user_prompt = input
.messages
.iter()
.rev()
.find_map(|m| match m {
ChatMessage::User { content, .. } => Some(content.clone()),
_ => None,
})
.unwrap_or_default();
let prompt = user_prompt.trim();
let (id, name, args) = if let Some(path) = prompt.strip_prefix("read ") {
("tc_read_1", "read", json!({"path": path.trim()}))
} else if let Some(rest) = prompt.strip_prefix("write ") {
let (path, content) = rest.split_once(' ').unwrap_or((rest, ""));
(
"tc_write_1",
"write",
json!({"path": path.trim(), "content": content}),
)
} else {
("tc_bash_1", "bash", json!({"command": prompt}))
};
vec![
ModelChunk::TextDelta {
msg_id: "scripted_msg".into(),
delta: format!("native model selected tool: {name}"),
},
ModelChunk::ToolCallStart {
id: id.into(),
name: name.into(),
},
ModelChunk::ToolCallEnd {
id: id.into(),
input: Some(args),
},
ModelChunk::Done {
stop_reason: "end_turn".into(),
usage: None,
},
]
}
#[derive(Debug, Clone)]
pub struct AnthropicConfig {
pub base_url: String,
pub api_key: String,
pub model: ResolvedModelConfig,
pub anthropic_version: String,
}
impl AnthropicConfig {
pub const DEFAULT_VERSION: &'static str = "2023-06-01";
}
#[derive(Debug, Clone)]
pub struct AnthropicModelClient {
http: reqwest::Client,
config: AnthropicConfig,
}
impl AnthropicModelClient {
pub fn new(config: AnthropicConfig) -> Self {
assert_eq!(
config.model.wire_protocol,
WireProtocol::Anthropic,
"resolved model protocol must match AnthropicModelClient"
);
let http = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(15))
.build()
.unwrap_or_else(|_| reqwest::Client::new());
Self { http, config }
}
fn endpoint(&self) -> String {
let base = self.config.base_url.trim_end_matches('/');
if base.ends_with("/messages") {
base.to_string()
} else {
format!("{base}/messages")
}
}
fn request_body(&self, input: &ModelTurnInput) -> Value {
let messages = chat_messages_to_anthropic_messages(&input.messages);
let tools = if matches!(input.tool_choice, ToolChoice::None) {
Vec::new()
} else {
let mut tools = input
.tools
.iter()
.map(tool_spec_to_anthropic_tool)
.collect::<Vec<_>>();
tools.extend(input.hosted_tools.iter().map(hosted_tool_to_anthropic_tool));
tools
};
let system_field = anthropic_system_field(input.system_prompt.as_deref());
let cached = apply_anthropic_cache_strategy(system_field, tools, messages);
let mut body = json!({
"model": self.config.model.model,
"max_tokens": self.config.model.max_output_tokens,
"messages": cached.messages,
"stream": true,
});
if let Some(sys) = cached.system {
body["system"] = sys;
}
if !cached.tools.is_empty() {
body["tools"] = json!(cached.tools);
if !matches!(input.tool_choice, ToolChoice::Auto) {
body["tool_choice"] = anthropic_tool_choice_value(&input.tool_choice);
}
}
if let Some(t) = self.config.model.temperature {
body["temperature"] = json!(t);
}
apply_anthropic_reasoning(&mut body, &self.config.model);
body
}
}
fn apply_anthropic_reasoning(body: &mut Value, model: &ResolvedModelConfig) {
if matches!(model.reasoning.mode, ReasoningMode::Default) {
return;
}
if matches!(model.reasoning.mode, ReasoningMode::Disabled) {
body["thinking"] = json!({"type": "disabled"});
return;
}
if let Some(tokens) = model.reasoning.budget_tokens {
body["thinking"] = json!({"type": "enabled", "budget_tokens": tokens});
return;
}
if let Some(effort) = model.reasoning.effort.as_deref() {
body["thinking"] = json!({"type": "adaptive"});
body["output_config"] = json!({"effort": effort});
return;
}
body["thinking"] = json!({"type": "enabled"});
}
fn anthropic_tool_choice_value(c: &ToolChoice) -> Value {
match c {
ToolChoice::Auto => json!({"type": "auto"}),
ToolChoice::None => json!({"type": "auto"}), ToolChoice::Required => json!({"type": "any"}),
ToolChoice::Tool(name) => json!({"type": "tool", "name": name}),
}
}
#[async_trait]
impl ModelClient for AnthropicModelClient {
fn hosted_capability(&self, capability: HostedCapability) -> CapabilitySupport {
match capability {
HostedCapability::WebSearch => {
official_endpoint_support(&self.config.base_url, &["api.anthropic.com"])
}
}
}
async fn stream(
&self,
input: ModelTurnInput,
) -> Result<BoxStream<'static, Result<ModelChunk, ModelClientError>>, ModelClientError> {
let resp = match self
.http
.post(self.endpoint())
.header("x-api-key", &self.config.api_key)
.header("anthropic-version", &self.config.anthropic_version)
.header("content-type", "application/json")
.json(&self.request_body(&input))
.send()
.await
{
Ok(r) => r,
Err(e) => return Err(classify_reqwest_error(&e, e.to_string())),
};
let status = resp.status();
if !status.is_success() {
let body_text = resp.text().await.unwrap_or_default();
return Err(classify_anthropic_http_error(status, &body_text));
}
let event_stream = resp.bytes_stream().eventsource();
let (tx, rx) = tokio::sync::mpsc::channel::<Result<ModelChunk, ModelClientError>>(8);
tokio::spawn(async move {
let mut state = AnthropicStreamState::default();
futures::pin_mut!(event_stream);
while let Some(ev) = event_stream.next().await {
let chunks = match ev {
Ok(event) => match state.feed_event(&event.event, &event.data) {
Ok(c) => c,
Err(e) => {
let _ = tx.send(Err(e)).await;
return;
}
},
Err(e) => {
let _ = tx
.send(Err(ModelClientError::Network(format!(
"SSE transport error: {e}"
))))
.await;
return;
}
};
for c in chunks {
if tx.send(Ok(c)).await.is_err() {
return;
}
}
}
if let Some(done) = state.finalize() {
let _ = tx.send(Ok(done)).await;
}
});
Ok(tokio_stream::wrappers::ReceiverStream::new(rx).boxed())
}
}
fn chat_messages_to_anthropic_messages(messages: &[ChatMessage]) -> Vec<Value> {
let mut out: Vec<Value> = Vec::with_capacity(messages.len());
let mut pending_tool_results: Vec<Value> = Vec::new();
let flush_tool_results = |bucket: &mut Vec<Value>, out: &mut Vec<Value>| {
if !bucket.is_empty() {
let blocks = std::mem::take(bucket);
out.push(json!({"role": "user", "content": blocks}));
}
};
for msg in messages {
match msg {
ChatMessage::User {
content,
attachments,
} => {
let mut blocks: Vec<Value> = std::mem::take(&mut pending_tool_results);
if !content.is_empty() {
blocks.push(json!({"type":"text","text":content}));
}
for att in attachments {
match att {
UserAttachment::Image(src) => {
blocks.push(image_to_anthropic_block(src));
}
}
}
if blocks.is_empty() {
blocks.push(json!({"type":"text","text":""}));
}
out.push(json!({"role": "user", "content": blocks}));
}
ChatMessage::Assistant {
text,
tool_calls,
thinking,
usage: _,
} => {
flush_tool_results(&mut pending_tool_results, &mut out);
let mut blocks: Vec<Value> = Vec::new();
if let Some(t) = thinking {
let mut tb = json!({"type": "thinking", "thinking": t.text});
if let Some(sig) = t.signature.as_deref() {
if !sig.is_empty() {
tb["signature"] = json!(sig);
}
}
blocks.push(tb);
}
if let Some(t) = text.as_deref() {
if !t.is_empty() {
blocks.push(json!({"type": "text", "text": t}));
}
}
for tc in tool_calls {
blocks.push(json!({
"type": "tool_use",
"id": tc.id,
"name": tc.name,
"input": tc.input,
}));
}
if blocks.is_empty() {
continue;
}
out.push(json!({"role": "assistant", "content": blocks}));
}
ChatMessage::Tool {
tool_call_id,
content,
is_error,
attachments,
} => {
let mut blocks: Vec<Value> = Vec::new();
if !content.is_empty() {
let replay = compact_tool_result_for_replay(content);
blocks.push(json!({"type": "text", "text": replay}));
}
for att in attachments {
let UserAttachment::Image(src) = att;
blocks.push(image_to_anthropic_block(src));
}
if blocks.is_empty() {
blocks.push(json!({"type": "text", "text": ""}));
}
pending_tool_results.push(json!({
"type": "tool_result",
"tool_use_id": tool_call_id,
"content": blocks,
"is_error": is_error,
}));
}
}
}
flush_tool_results(&mut pending_tool_results, &mut out);
out
}
fn anthropic_system_field(prompt: Option<&str>) -> Option<Value> {
let s = prompt?.trim();
if s.is_empty() {
return None;
}
Some(json!([{"type": "text", "text": s}]))
}
fn image_to_anthropic_block(src: &ImageSource) -> Value {
let source = match &src.data {
ImageData::Base64(b64) => json!({
"type": "base64",
"media_type": src.media_type,
"data": b64,
}),
ImageData::Url(url) => json!({
"type": "url",
"url": url,
}),
};
json!({"type": "image", "source": source})
}
fn tool_spec_to_anthropic_tool(spec: &ToolSpec) -> Value {
json!({
"name": spec.name,
"description": spec.description,
"input_schema": spec.input_schema,
})
}
fn hosted_tool_to_anthropic_tool(tool: &HostedTool) -> Value {
match tool {
HostedTool::WebSearch => json!({
"type": "web_search_20250305",
"name": "web_search",
}),
}
}
struct AnthropicCached {
system: Option<Value>,
tools: Vec<Value>,
messages: Vec<Value>,
}
fn apply_anthropic_cache_strategy(
system: Option<Value>,
tools: Vec<Value>,
messages: Vec<Value>,
) -> AnthropicCached {
let mut system = system;
if let Some(sys) = system.as_mut() {
if let Some(arr) = sys.as_array_mut() {
if let Some(last) = arr.last_mut() {
if last
.get("text")
.and_then(|v| v.as_str())
.map(|s| !s.is_empty())
.unwrap_or(false)
{
last["cache_control"] = json!({"type": "ephemeral"});
}
}
}
}
let mut tools = tools;
if let Some(last) = tools.last_mut() {
last["cache_control"] = json!({"type": "ephemeral"});
}
let mut messages = messages;
if let Some(last) = messages.last_mut() {
if let Some(blocks) = last.get_mut("content").and_then(|v| v.as_array_mut()) {
if let Some(last_block) = blocks.last_mut() {
last_block["cache_control"] = json!({"type": "ephemeral"});
}
}
}
if messages.len() > 30 {
let mid = messages.len() / 2;
if let Some(blocks) = messages[mid]
.get_mut("content")
.and_then(|v| v.as_array_mut())
{
if let Some(last_block) = blocks.last_mut() {
last_block["cache_control"] = json!({"type": "ephemeral"});
}
}
}
AnthropicCached {
system,
tools,
messages,
}
}
#[derive(Debug, Default)]
struct AnthropicStreamState {
msg_id: Option<String>,
blocks: std::collections::HashMap<u64, AnthropicBlock>,
stop_reason: Option<String>,
pending_usage: Option<HarnessUsage>,
done_emitted: bool,
}
#[derive(Debug)]
enum AnthropicBlock {
Text,
Thinking { thinking_id: String },
ToolUse { id: String },
Ignored,
}
impl AnthropicStreamState {
fn feed_event(&mut self, event: &str, data: &str) -> Result<Vec<ModelChunk>, ModelClientError> {
match event {
"ping" | "" => return Ok(vec![]),
"error" => {
return Err(ModelClientError::Other(format!(
"anthropic stream error event: {data}"
)));
}
_ => {}
}
let value: Value = serde_json::from_str(data).map_err(|e| {
ModelClientError::Other(format!(
"anthropic SSE data not JSON (event={event}): {e}; raw={data}"
))
})?;
let mut out: Vec<ModelChunk> = Vec::new();
match event {
"message_start" => {
let msg = value.get("message");
if let Some(id) = msg.and_then(|m| m.get("id")).and_then(|v| v.as_str()) {
if !id.is_empty() {
self.msg_id = Some(id.to_string());
}
}
if let Some(u) = msg.and_then(|m| m.get("usage")) {
self.pending_usage =
Some(merge_anthropic_usage(self.pending_usage.clone(), u, true));
}
}
"content_block_start" => {
let index = value.get("index").and_then(|v| v.as_u64()).unwrap_or(0);
let block = value.get("content_block");
let kind = block.and_then(|b| b.get("type")).and_then(|v| v.as_str());
match kind {
Some("text") => {
self.blocks.insert(index, AnthropicBlock::Text);
}
Some("thinking") => {
let thinking_id = self
.msg_id
.clone()
.map(|m| format!("{m}_t{index}"))
.unwrap_or_else(|| format!("thinking_{index}"));
self.blocks
.insert(index, AnthropicBlock::Thinking { thinking_id });
}
Some("tool_use") => {
let id = block
.and_then(|b| b.get("id"))
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
let name = block
.and_then(|b| b.get("name"))
.and_then(|v| v.as_str())
.unwrap_or_default()
.to_string();
if !id.is_empty() && !name.is_empty() {
out.push(ModelChunk::ToolCallStart {
id: id.clone(),
name,
});
}
self.blocks.insert(index, AnthropicBlock::ToolUse { id });
}
Some("server_tool_use") | Some("web_search_tool_result") => {
self.blocks.insert(index, AnthropicBlock::Ignored);
}
_ => {
self.blocks.insert(index, AnthropicBlock::Ignored);
}
}
}
"content_block_delta" => {
let index = value.get("index").and_then(|v| v.as_u64()).unwrap_or(0);
let delta = match value.get("delta") {
Some(d) => d,
None => return Ok(out),
};
let delta_type = delta.get("type").and_then(|v| v.as_str()).unwrap_or("");
match (self.blocks.get(&index), delta_type) {
(Some(AnthropicBlock::Text), "text_delta") => {
if let Some(text) = delta.get("text").and_then(|v| v.as_str()) {
if !text.is_empty() {
let msg_id = self
.msg_id
.clone()
.unwrap_or_else(|| "msg_anthropic_default".into());
out.push(ModelChunk::TextDelta {
msg_id,
delta: text.to_string(),
});
}
}
}
(Some(AnthropicBlock::Thinking { thinking_id }), "thinking_delta") => {
if let Some(text) = delta.get("thinking").and_then(|v| v.as_str()) {
if !text.is_empty() {
out.push(ModelChunk::ThinkingDelta {
thinking_id: thinking_id.clone(),
delta: text.to_string(),
signature: None,
});
}
}
}
(Some(AnthropicBlock::Thinking { thinking_id }), "signature_delta") => {
if let Some(sig) = delta.get("signature").and_then(|v| v.as_str()) {
out.push(ModelChunk::ThinkingDelta {
thinking_id: thinking_id.clone(),
delta: String::new(),
signature: Some(sig.to_string()),
});
}
}
(Some(AnthropicBlock::ToolUse { id }), "input_json_delta") => {
if let Some(partial) = delta.get("partial_json").and_then(|v| v.as_str()) {
if !partial.is_empty() {
out.push(ModelChunk::ToolCallInputDelta {
id: id.clone(),
delta: partial.to_string(),
});
}
}
}
_ => { }
}
}
"content_block_stop" => {
let index = value.get("index").and_then(|v| v.as_u64()).unwrap_or(0);
if let Some(AnthropicBlock::ToolUse { id }) = self.blocks.get(&index) {
out.push(ModelChunk::ToolCallEnd {
id: id.clone(),
input: None,
});
}
}
"message_delta" => {
if let Some(reason) = value
.get("delta")
.and_then(|d| d.get("stop_reason"))
.and_then(|v| v.as_str())
{
self.stop_reason = Some(reason.to_string());
}
if let Some(u) = value.get("usage") {
self.pending_usage =
Some(merge_anthropic_usage(self.pending_usage.clone(), u, false));
}
}
"message_stop" => {
if let Some(done) = self.emit_done() {
out.push(done);
}
}
_ => { }
}
Ok(out)
}
fn finalize(&mut self) -> Option<ModelChunk> {
self.emit_done()
}
fn emit_done(&mut self) -> Option<ModelChunk> {
if self.done_emitted {
return None;
}
self.done_emitted = true;
let stop_reason = map_anthropic_stop_reason(self.stop_reason.as_deref());
Some(ModelChunk::Done {
stop_reason,
usage: self.pending_usage.take(),
})
}
}
fn merge_anthropic_usage(
prior: Option<HarnessUsage>,
incoming: &Value,
include_input: bool,
) -> HarnessUsage {
let mut u = prior.unwrap_or_default();
if include_input {
if let Some(v) = incoming.get("input_tokens").and_then(|v| v.as_u64()) {
u.input_tokens = v;
}
if let Some(v) = incoming
.get("cache_read_input_tokens")
.and_then(|v| v.as_u64())
{
u.cache_read_input_tokens = v;
}
if let Some(v) = incoming
.get("cache_creation_input_tokens")
.and_then(|v| v.as_u64())
{
u.cache_creation_input_tokens = v;
}
}
if let Some(v) = incoming.get("output_tokens").and_then(|v| v.as_u64()) {
u.output_tokens = v;
}
u
}
fn map_anthropic_stop_reason(reason: Option<&str>) -> String {
match reason {
Some("end_turn") | Some("stop_sequence") | Some("tool_use") => "end_turn".into(),
Some("max_tokens") => "max_tokens".into(),
Some("refusal") => "refusal".into(),
Some(other) if !other.is_empty() => other.to_string(),
_ => "end_turn".into(),
}
}
fn classify_anthropic_http_error(status: reqwest::StatusCode, body: &str) -> ModelClientError {
use reqwest::StatusCode;
let snippet = body.chars().take(512).collect::<String>();
if status == StatusCode::TOO_MANY_REQUESTS {
return ModelClientError::RateLimit(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::UNAUTHORIZED || status == StatusCode::FORBIDDEN {
return ModelClientError::Auth(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::BAD_REQUEST && looks_like_context_overflow(body) {
return ModelClientError::ContextOverflow(format!("HTTP {status}: {snippet}"));
}
if status == StatusCode::BAD_REQUEST {
return ModelClientError::BadRequest(format!("HTTP {status}: {snippet}"));
}
if status.is_server_error() {
return ModelClientError::ServerError(format!("HTTP {status}: {snippet}"));
}
ModelClientError::Other(format!("HTTP {status}: {snippet}"))
}
#[derive(Debug, Clone)]
pub struct OpenAiResponsesConfig {
pub base_url: String,
pub api_key: String,
pub model: ResolvedModelConfig,
pub reasoning_summary: Option<String>,
}
impl OpenAiResponsesConfig {
pub const DEFAULT_BASE_URL: &'static str = "https://api.openai.com/v1";
}
#[derive(Debug, Clone)]
pub struct OpenAiResponsesModelClient {
http: reqwest::Client,
config: OpenAiResponsesConfig,
}
impl OpenAiResponsesModelClient {
pub fn new(config: OpenAiResponsesConfig) -> Self {
assert_eq!(
config.model.wire_protocol,
WireProtocol::OpenAiResponses,
"resolved model protocol must match OpenAiResponsesModelClient"
);
let http = reqwest::Client::builder()
.connect_timeout(std::time::Duration::from_secs(15))
.build()
.unwrap_or_else(|_| reqwest::Client::new());
Self { http, config }
}
fn endpoint(&self) -> String {
let base = self.config.base_url.trim_end_matches('/');
if base.ends_with("/responses") {
base.to_string()
} else {
format!("{base}/responses")
}
}
fn request_body(&self, input: &ModelTurnInput) -> Value {
let mut body = json!({
"model": self.config.model.model,
"input": chat_messages_to_responses_input(&input.messages),
"stream": true,
"store": false,
});
if let Some(sys) = input.system_prompt.as_deref().filter(|s| !s.is_empty()) {
body["instructions"] = json!(sys);
}
let advertise_tools = !matches!(input.tool_choice, ToolChoice::None);
let mut tools: Vec<Value> = Vec::new();
if advertise_tools {
tools.extend(input.tools.iter().map(tool_spec_to_responses_tool));
tools.extend(input.hosted_tools.iter().map(hosted_tool_to_responses_tool));
}
if !tools.is_empty() {
body["tools"] = json!(tools);
let send_choice = match input.tool_choice {
ToolChoice::Auto | ToolChoice::Required => true,
ToolChoice::Tool(_) => !input.tools.is_empty(),
ToolChoice::None => false, };
if send_choice {
body["tool_choice"] = responses_tool_choice_value(&input.tool_choice);
}
if let Some(parallel) = input.parallel_tool_calls {
body["parallel_tool_calls"] = json!(parallel);
}
}
if let Some(temperature) = self.config.model.temperature {
body["temperature"] = json!(temperature);
}
body["max_output_tokens"] = json!(self.config.model.max_output_tokens);
let effort = self
.config
.model
.reasoning
.effort
.as_deref()
.filter(|s| !s.is_empty());
let summary = self
.config
.reasoning_summary
.as_deref()
.filter(|s| !s.is_empty());
if effort.is_some() || summary.is_some() {
let mut reasoning = json!({});
if let Some(e) = effort {
reasoning["effort"] = json!(e);
}
if let Some(s) = summary {
reasoning["summary"] = json!(s);
}
body["reasoning"] = reasoning;
}
body["include"] = json!(["reasoning.encrypted_content"]);
body
}
}
fn tool_spec_to_responses_tool(spec: &ToolSpec) -> Value {
json!({
"type": "function",
"name": spec.name,
"description": spec.description,
"parameters": spec.input_schema,
})
}
fn hosted_tool_to_responses_tool(tool: &HostedTool) -> Value {
match tool {
HostedTool::WebSearch => json!({ "type": "web_search" }),
}
}
fn responses_tool_choice_value(c: &ToolChoice) -> Value {
match c {
ToolChoice::Auto => json!("auto"),
ToolChoice::None => json!("none"),
ToolChoice::Required => json!("required"),
ToolChoice::Tool(name) => json!({ "type": "function", "name": name }),
}
}
fn image_to_responses_data_url(src: &ImageSource) -> String {
match &src.data {
ImageData::Base64(b64) => format!("data:{};base64,{}", src.media_type, b64),
ImageData::Url(u) => u.clone(),
}
}
fn encode_reasoning_signature(item_id: &str, encrypted_content: &str) -> String {
format!("{item_id}\n{encrypted_content}")
}
fn decode_reasoning_signature(sig: &str) -> Option<(String, String)> {
let (id, enc) = sig.split_once('\n')?;
if id.is_empty() || enc.is_empty() {
return None;
}
Some((id.to_string(), enc.to_string()))
}
fn chat_messages_to_responses_input(messages: &[ChatMessage]) -> Vec<Value> {
let mut out: Vec<Value> = Vec::with_capacity(messages.len());
for msg in messages {
match msg {
ChatMessage::User {
content,
attachments,
} => {
let mut parts: Vec<Value> = Vec::with_capacity(attachments.len() + 1);
if !content.is_empty() {
parts.push(json!({"type": "input_text", "text": content}));
}
for att in attachments {
let UserAttachment::Image(src) = att;
parts.push(json!({
"type": "input_image",
"image_url": image_to_responses_data_url(src),
}));
}
if parts.is_empty() {
parts.push(json!({"type": "input_text", "text": ""}));
}
out.push(json!({"role": "user", "content": parts}));
}
ChatMessage::Assistant {
text,
tool_calls,
thinking,
usage: _,
} => {
if let Some(t) = thinking {
if let Some((item_id, enc)) =
t.signature.as_deref().and_then(decode_reasoning_signature)
{
let summary = if t.text.is_empty() {
json!([])
} else {
json!([{"type": "summary_text", "text": t.text}])
};
out.push(json!({
"type": "reasoning",
"id": item_id,
"summary": summary,
"encrypted_content": enc,
}));
}
}
if let Some(t) = text.as_deref().filter(|s| !s.is_empty()) {
out.push(json!({
"role": "assistant",
"content": [{"type": "output_text", "text": t}],
}));
}
for tc in tool_calls {
out.push(json!({
"type": "function_call",
"call_id": tc.id,
"name": tc.name,
"arguments": tool_invocation_args_for_wire(tc),
}));
}
}
ChatMessage::Tool {
tool_call_id,
content,
is_error: _,
attachments,
} => {
if attachments.is_empty() {
let replay = compact_tool_result_for_replay(content).into_owned();
out.push(json!({
"type": "function_call_output",
"call_id": tool_call_id,
"output": replay,
}));
} else {
let mut parts: Vec<Value> = Vec::new();
if !content.is_empty() {
let replay = compact_tool_result_for_replay(content).into_owned();
parts.push(json!({"type": "input_text", "text": replay}));
}
for att in attachments {
let UserAttachment::Image(src) = att;
parts.push(json!({
"type": "input_image",
"image_url": image_to_responses_data_url(src),
}));
}
out.push(json!({
"type": "function_call_output",
"call_id": tool_call_id,
"output": parts,
}));
}
}
}
}
out
}
fn parse_responses_usage(usage: Option<&Value>) -> Option<HarnessUsage> {
let u = usage?;
let input = u.get("input_tokens").and_then(|v| v.as_u64()).unwrap_or(0);
let output = u.get("output_tokens").and_then(|v| v.as_u64()).unwrap_or(0);
let cache_read = u
.get("input_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(|v| v.as_u64())
.unwrap_or(0);
if input == 0 && output == 0 && cache_read == 0 {
return None;
}
Some(HarnessUsage {
input_tokens: input,
output_tokens: output,
cache_read_input_tokens: cache_read,
cache_creation_input_tokens: 0,
compaction_input_tokens: 0,
compaction_output_tokens: 0,
})
}
fn map_responses_stop_reason(reason: Option<&str>) -> String {
match reason {
Some("max_output_tokens") => "max_tokens".into(),
Some("content_filter") => "refusal".into(),
_ => "end_turn".into(),
}
}
fn classify_responses_error(code: &str, message: &str) -> ModelClientError {
let full = if code.is_empty() {
message.to_string()
} else {
format!("{code}: {message}")
};
if code == "context_length_exceeded" || looks_like_context_overflow(message) {
return ModelClientError::ContextOverflow(full);
}
if code == "rate_limit_exceeded" {
return ModelClientError::RateLimit(full);
}
ModelClientError::BadRequest(full)
}
#[async_trait]
impl ModelClient for OpenAiResponsesModelClient {
fn hosted_capability(&self, capability: HostedCapability) -> CapabilitySupport {
match capability {
HostedCapability::WebSearch => {
official_endpoint_support(&self.config.base_url, &["api.openai.com"])
}
}
}
async fn stream(
&self,
input: ModelTurnInput,
) -> Result<BoxStream<'static, Result<ModelChunk, ModelClientError>>, ModelClientError> {
let body = self.request_body(&input);
let resp = match self
.http
.post(self.endpoint())
.bearer_auth(&self.config.api_key)
.json(&body)
.send()
.await
{
Ok(r) => r,
Err(e) => return Err(classify_reqwest_error(&e, e.to_string())),
};
let status = resp.status();
if !status.is_success() {
let body_text = resp.text().await.unwrap_or_default();
return Err(classify_openai_http_error(status, &body_text));
}
let event_stream = resp.bytes_stream().eventsource();
let (tx, rx) = tokio::sync::mpsc::channel::<Result<ModelChunk, ModelClientError>>(8);
tokio::spawn(async move {
let mut state = OpenAiResponsesStreamState::default();
futures::pin_mut!(event_stream);
while let Some(ev) = event_stream.next().await {
let chunks = match ev {
Ok(event) => match state.feed_data(&event.data) {
Ok(c) => c,
Err(e) => {
let _ = tx.send(Err(e)).await;
return;
}
},
Err(e) => {
let _ = tx
.send(Err(ModelClientError::Network(format!(
"SSE transport error: {e}"
))))
.await;
return;
}
};
for c in chunks {
if tx.send(Ok(c)).await.is_err() {
return;
}
}
}
if state.ended_cleanly() {
if let Some(done) = state.finalize() {
let _ = tx.send(Ok(done)).await;
}
} else {
let _ = tx
.send(Err(ModelClientError::Network(
"model stream closed before completion (no terminal response event) \
— connection dropped or upstream truncated the response"
.into(),
)))
.await;
}
});
Ok(tokio_stream::wrappers::ReceiverStream::new(rx).boxed())
}
}
fn official_endpoint_support(base_url: &str, official_hosts: &[&str]) -> CapabilitySupport {
let Ok(url) = reqwest::Url::parse(base_url) else {
return CapabilitySupport::Unknown;
};
let Some(host) = url.host_str() else {
return CapabilitySupport::Unknown;
};
if official_hosts
.iter()
.any(|official| host.eq_ignore_ascii_case(official))
{
CapabilitySupport::Supported
} else {
CapabilitySupport::Unknown
}
}
#[derive(Debug, Default)]
struct OpenAiResponsesStreamState {
response_id: Option<String>,
fc_call_by_item: std::collections::HashMap<String, String>,
summary_parts_seen: std::collections::HashSet<String>,
stop_reason: Option<String>,
pending_usage: Option<HarnessUsage>,
done_emitted: bool,
saw_terminal: bool,
}
impl OpenAiResponsesStreamState {
fn feed_data(&mut self, data: &str) -> Result<Vec<ModelChunk>, ModelClientError> {
let trimmed = data.trim();
if trimmed.is_empty() || trimmed == "[DONE]" {
return Ok(vec![]);
}
let value: Value = serde_json::from_str(trimmed).map_err(|e| {
ModelClientError::Other(format!("Responses SSE data not JSON: {e}; raw={trimmed}"))
})?;
let event_type = value.get("type").and_then(|v| v.as_str()).unwrap_or("");
if let Some(rid) = value
.get("response")
.and_then(|r| r.get("id"))
.and_then(|v| v.as_str())
{
if self.response_id.is_none() && !rid.is_empty() {
self.response_id = Some(rid.to_string());
}
}
let mut out: Vec<ModelChunk> = Vec::new();
match event_type {
"response.output_text.delta" => {
if let Some(delta) = value.get("delta").and_then(|v| v.as_str()) {
if !delta.is_empty() {
let msg_id = value
.get("item_id")
.and_then(|v| v.as_str())
.map(String::from)
.or_else(|| self.response_id.clone())
.unwrap_or_else(|| "msg_responses_default".into());
out.push(ModelChunk::TextDelta {
msg_id,
delta: delta.to_string(),
});
}
}
}
"response.reasoning_summary_text.delta" | "response.reasoning_summary.delta" => {
if let Some(delta) = value.get("delta").and_then(|v| v.as_str()) {
if !delta.is_empty() {
let item_id = value
.get("item_id")
.and_then(|v| v.as_str())
.unwrap_or("reasoning-0");
let idx = value
.get("summary_index")
.and_then(|v| v.as_u64())
.unwrap_or(0);
let first_of_part =
self.summary_parts_seen.insert(format!("{item_id}:{idx}"));
let text = if first_of_part && idx > 0 {
format!("\n\n{delta}")
} else {
delta.to_string()
};
out.push(ModelChunk::ThinkingDelta {
thinking_id: item_id.to_string(),
delta: text,
signature: None,
});
}
}
}
"response.reasoning_text.delta" => {
if let Some(delta) = value.get("delta").and_then(|v| v.as_str()) {
if !delta.is_empty() {
let thinking_id = value
.get("item_id")
.and_then(|v| v.as_str())
.unwrap_or("reasoning-0")
.to_string();
out.push(ModelChunk::ThinkingDelta {
thinking_id,
delta: delta.to_string(),
signature: None,
});
}
}
}
"response.output_item.added" => {
if let Some(item) = value.get("item") {
if item.get("type").and_then(|v| v.as_str()) == Some("function_call") {
let fc_id = item.get("id").and_then(|v| v.as_str()).unwrap_or("");
let call_id = item
.get("call_id")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.unwrap_or(fc_id);
let name = item
.get("name")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
if !call_id.is_empty() {
if !fc_id.is_empty() {
self.fc_call_by_item
.insert(fc_id.to_string(), call_id.to_string());
}
out.push(ModelChunk::ToolCallStart {
id: call_id.to_string(),
name,
});
}
}
}
}
"response.function_call_arguments.delta" => {
let item_id = value.get("item_id").and_then(|v| v.as_str()).unwrap_or("");
if let Some(delta) = value.get("delta").and_then(|v| v.as_str()) {
if !delta.is_empty() {
let call_id = self
.fc_call_by_item
.get(item_id)
.cloned()
.unwrap_or_else(|| item_id.to_string());
out.push(ModelChunk::ToolCallInputDelta {
id: call_id,
delta: delta.to_string(),
});
}
}
}
"response.output_item.done" => {
if let Some(item) = value.get("item") {
let itype = item.get("type").and_then(|v| v.as_str()).unwrap_or("");
if itype == "function_call" {
let fc_id = item.get("id").and_then(|v| v.as_str()).unwrap_or("");
let call_id = self
.fc_call_by_item
.get(fc_id)
.cloned()
.or_else(|| {
item.get("call_id")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
.map(String::from)
})
.unwrap_or_else(|| fc_id.to_string());
if !fc_id.is_empty() && !self.fc_call_by_item.contains_key(fc_id) {
let name = item
.get("name")
.and_then(|v| v.as_str())
.unwrap_or("")
.to_string();
out.push(ModelChunk::ToolCallStart {
id: call_id.clone(),
name,
});
self.fc_call_by_item
.insert(fc_id.to_string(), call_id.clone());
}
let input = item
.get("arguments")
.and_then(|v| v.as_str())
.map(str::trim)
.filter(|s| !s.is_empty())
.and_then(|s| serde_json::from_str::<Value>(s).ok());
out.push(ModelChunk::ToolCallEnd { id: call_id, input });
} else if itype == "reasoning" {
let rid = item.get("id").and_then(|v| v.as_str()).unwrap_or("");
if let Some(enc) = item
.get("encrypted_content")
.and_then(|v| v.as_str())
.filter(|s| !s.is_empty())
{
if !rid.is_empty() {
out.push(ModelChunk::ThinkingDelta {
thinking_id: rid.to_string(),
delta: String::new(),
signature: Some(encode_reasoning_signature(rid, enc)),
});
}
}
}
}
}
"response.completed" | "response.incomplete" => {
self.saw_terminal = true;
let resp = value.get("response");
self.pending_usage = parse_responses_usage(resp.and_then(|r| r.get("usage")));
let reason = resp
.and_then(|r| r.get("incomplete_details"))
.and_then(|d| d.get("reason"))
.and_then(|v| v.as_str());
self.stop_reason = Some(map_responses_stop_reason(reason));
if let Some(done) = self.emit_done() {
out.push(done);
}
}
"response.failed" => {
self.saw_terminal = true;
let err = value.get("response").and_then(|r| r.get("error"));
let code = err
.and_then(|e| e.get("code"))
.and_then(|v| v.as_str())
.unwrap_or("");
let message = err
.and_then(|e| e.get("message"))
.and_then(|v| v.as_str())
.unwrap_or("Responses response failed");
return Err(classify_responses_error(code, message));
}
"error" => {
self.saw_terminal = true;
let code = value.get("code").and_then(|v| v.as_str()).unwrap_or("");
let message = value
.get("message")
.and_then(|v| v.as_str())
.unwrap_or("Responses stream error");
return Err(classify_responses_error(code, message));
}
_ => {}
}
Ok(out)
}
fn finalize(&mut self) -> Option<ModelChunk> {
self.emit_done()
}
fn ended_cleanly(&self) -> bool {
self.saw_terminal || self.done_emitted
}
fn emit_done(&mut self) -> Option<ModelChunk> {
if self.done_emitted {
return None;
}
self.done_emitted = true;
Some(ModelChunk::Done {
stop_reason: self
.stop_reason
.clone()
.unwrap_or_else(|| "end_turn".into()),
usage: self.pending_usage.take(),
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model_catalog::{
LimitsSource, ModelCapabilities, ModelLimits, ReasoningConfig, ReasoningOption,
};
fn resolved_model(
id: &str,
protocol: WireProtocol,
max_output_tokens: u64,
temperature: Option<f64>,
reasoning: ReasoningConfig,
) -> ResolvedModelConfig {
ResolvedModelConfig {
model: id.into(),
wire_protocol: protocol,
max_output_tokens,
temperature,
reasoning,
capabilities: ModelCapabilities {
id: id.into(),
limits: ModelLimits {
context: 1_000_000,
input: None,
output: 384_000,
},
reasoning: true,
reasoning_options: vec![
ReasoningOption::Toggle,
ReasoningOption::Effort {
values: vec!["none".into(), "low".into(), "medium".into(), "high".into()],
},
],
temperature: true,
tool_call: true,
interleaved: None,
status: None,
limits_source: LimitsSource::Catalog,
},
}
}
fn default_model(id: &str, protocol: WireProtocol) -> ResolvedModelConfig {
resolved_model(id, protocol, 2_048, None, ReasoningConfig::default())
}
fn user(prompt: &str) -> ModelTurnInput {
ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: prompt.into(),
attachments: vec![],
}],
tools: vec![],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
}
}
fn bash_spec() -> ToolSpec {
ToolSpec {
name: "bash".into(),
description: "Run a shell command inside the sandbox.".into(),
input_schema: json!({
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
"additionalProperties": false
}),
}
}
#[test]
fn openai_client_builds_chat_completions_request() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test/v1/".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
assert_eq!(
client.endpoint(),
"https://example.test/v1/chat/completions"
);
let client_with_v1 = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
assert_eq!(
client_with_v1.endpoint(),
"https://example.test/v1/chat/completions"
);
let glm = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://open.bigmodel.cn/api/coding/paas/v4".into(),
api_key: "sk-test".into(),
model: default_model("glm-4.6", WireProtocol::OpenAiCompatible),
});
assert_eq!(
glm.endpoint(),
"https://open.bigmodel.cn/api/coding/paas/v4/chat/completions"
);
let body = client.request_body(&user("hello"));
assert_eq!(body["model"], "gpt-test");
assert_eq!(body["messages"][0]["role"], "user");
assert_eq!(body["messages"][0]["content"], "hello");
assert!(body.get("tools").is_none());
assert!(body.get("tool_choice").is_none());
let with_tools = ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "hello".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
};
let body = client.request_body(&with_tools);
assert_eq!(body["tools"][0]["function"]["name"], "bash");
assert_eq!(
body["tools"][0]["function"]["parameters"]["required"][0],
"command"
);
assert_eq!(body["tool_choice"], "auto");
assert!(body.get("parallel_tool_calls").is_none());
}
#[test]
fn openai_client_emits_tool_choice_required() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Required,
parallel_tool_calls: Some(false),
});
assert_eq!(body["tool_choice"], "required");
assert_eq!(body["parallel_tool_calls"], false);
}
#[test]
fn openai_client_emits_tool_choice_named_tool() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Tool("bash".into()),
parallel_tool_calls: None,
});
assert_eq!(body["tool_choice"]["type"], "function");
assert_eq!(body["tool_choice"]["function"]["name"], "bash");
}
#[test]
fn openai_client_drops_tools_when_choice_is_none() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::None,
parallel_tool_calls: None,
});
assert!(body.get("tools").is_none(), "tools should be dropped");
assert!(body.get("tool_choice").is_none());
}
#[tokio::test]
async fn openai_compatible_rejects_hosted_web_search() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
let err = match client
.stream(ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "search".into(),
attachments: vec![],
}],
tools: vec![],
hosted_tools: vec![HostedTool::WebSearch],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
})
.await
{
Ok(_) => panic!("expected hosted tool rejection"),
Err(err) => err,
};
assert!(err.to_string().contains("Responses API"));
}
#[test]
fn tool_choice_parse_handles_canonical_strings() {
assert!(matches!(ToolChoice::parse(""), ToolChoice::Auto));
assert!(matches!(ToolChoice::parse("auto"), ToolChoice::Auto));
assert!(matches!(ToolChoice::parse("AUTO"), ToolChoice::Auto));
assert!(matches!(ToolChoice::parse("none"), ToolChoice::None));
assert!(matches!(
ToolChoice::parse("required"),
ToolChoice::Required
));
assert!(matches!(ToolChoice::parse("any"), ToolChoice::Required));
match ToolChoice::parse("tool:bash") {
ToolChoice::Tool(name) => assert_eq!(name, "bash"),
other => panic!("expected Tool(bash), got {other:?}"),
}
assert!(matches!(ToolChoice::parse("garbage"), ToolChoice::Auto));
}
#[test]
fn openai_client_prepends_system_when_set() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
let input = ModelTurnInput {
system_prompt: Some("you are concise".into()),
messages: vec![ChatMessage::User {
content: "hi".into(),
attachments: vec![],
}],
tools: vec![],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
};
let body = client.request_body(&input);
assert_eq!(body["messages"][0]["role"], "system");
assert_eq!(body["messages"][0]["content"], "you are concise");
assert_eq!(body["messages"][1]["role"], "user");
}
#[test]
fn parse_openai_usage_extracts_token_counts() {
let u = parse_openai_usage(Some(&json!({
"prompt_tokens": 12,
"completion_tokens": 7,
"total_tokens": 19,
"prompt_tokens_details": {"cached_tokens": 4}
})))
.expect("usage parsed");
assert_eq!(u.input_tokens, 12);
assert_eq!(u.output_tokens, 7);
assert_eq!(u.cache_read_input_tokens, 4);
assert_eq!(u.cache_creation_input_tokens, 0);
}
#[test]
fn parse_openai_usage_without_cache_details() {
let u = parse_openai_usage(Some(&json!({
"prompt_tokens": 200,
"completion_tokens": 30
})))
.expect("usage parsed");
assert_eq!(u.input_tokens, 200);
assert_eq!(u.output_tokens, 30);
assert_eq!(u.cache_read_input_tokens, 0);
}
#[test]
fn openai_stream_state_emits_text_deltas_then_done() {
let mut state = OpenAiStreamState::default();
let out = state
.feed_data(
r#"{"id":"chatcmpl-1","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}"#,
)
.unwrap();
assert!(out.is_empty(), "empty content shouldn't emit");
let out = state
.feed_data(r#"{"choices":[{"index":0,"delta":{"content":"Hello"}}]}"#)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::TextDelta { msg_id, delta } => {
assert_eq!(msg_id, "chatcmpl-1");
assert_eq!(delta, "Hello");
}
other => panic!("expected TextDelta, got {other:?}"),
}
let out = state
.feed_data(r#"{"choices":[{"index":0,"delta":{"content":" world"}}]}"#)
.unwrap();
assert_eq!(out.len(), 1);
let out = state
.feed_data(r#"{"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}"#)
.unwrap();
assert!(out.is_empty());
let out = state
.feed_data(r#"{"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":3}}"#)
.unwrap();
assert!(out.is_empty());
let out = state.feed_data("[DONE]").unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::Done { stop_reason, usage } => {
assert_eq!(stop_reason, "end_turn");
let u = usage.as_ref().expect("usage propagated");
assert_eq!(u.input_tokens, 10);
assert_eq!(u.output_tokens, 3);
}
other => panic!("expected Done, got {other:?}"),
}
}
#[test]
fn openai_stream_state_emits_tool_call_chunks() {
let mut state = OpenAiStreamState::default();
let out = state
.feed_data(
r#"{"id":"c1","choices":[{"index":0,"delta":{"role":"assistant","tool_calls":[{"index":0,"id":"call_x","type":"function","function":{"name":"bash","arguments":""}}]}}]}"#,
)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::ToolCallStart { id, name } => {
assert_eq!(id, "call_x");
assert_eq!(name, "bash");
}
other => panic!("expected ToolCallStart, got {other:?}"),
}
let out = state
.feed_data(
r#"{"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"{\""}}]}}]}"#,
)
.unwrap();
assert_eq!(out.len(), 1);
let ModelChunk::ToolCallInputDelta { delta, .. } = &out[0] else {
panic!("expected ToolCallInputDelta");
};
assert_eq!(delta, "{\"");
let out = state
.feed_data(
r#"{"choices":[{"index":0,"delta":{"tool_calls":[{"index":0,"function":{"arguments":"cmd\":\"pwd\"}"}}]}}]}"#,
)
.unwrap();
let ModelChunk::ToolCallInputDelta { delta, .. } = &out[0] else {
panic!("expected ToolCallInputDelta");
};
assert_eq!(delta, "cmd\":\"pwd\"}");
let out = state
.feed_data(r#"{"choices":[{"index":0,"delta":{},"finish_reason":"tool_calls"}]}"#)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::ToolCallEnd { id, input } => {
assert_eq!(id, "call_x");
assert!(input.is_none(), "OpenAI streaming defers parsing");
}
other => panic!("expected ToolCallEnd, got {other:?}"),
}
let final_chunk = state.finalize().expect("finalize emits Done");
match final_chunk {
ModelChunk::Done { stop_reason, .. } => assert_eq!(stop_reason, "end_turn"),
other => panic!("expected Done from finalize, got {other:?}"),
}
}
#[tokio::test]
async fn collect_model_response_preserves_raw_tool_arguments() {
let chunks = vec![
Ok(ModelChunk::ToolCallStart {
id: "call_x".into(),
name: "bash".into(),
}),
Ok(ModelChunk::ToolCallInputDelta {
id: "call_x".into(),
delta: r#"{ "b": 2, "#.into(),
}),
Ok(ModelChunk::ToolCallInputDelta {
id: "call_x".into(),
delta: r#""a": 1 }"#.into(),
}),
Ok(ModelChunk::ToolCallEnd {
id: "call_x".into(),
input: None,
}),
Ok(ModelChunk::Done {
stop_reason: "end_turn".into(),
usage: None,
}),
];
let response = collect_model_response(futures::stream::iter(chunks).boxed())
.await
.unwrap();
let ModelResponse::ToolCall { invocation, .. } = response else {
panic!("expected tool call response");
};
assert_eq!(invocation.input, json!({"b": 2, "a": 1}));
assert_eq!(
invocation.raw_emitted_args.as_deref(),
Some(r#"{ "b": 2, "a": 1 }"#)
);
}
#[test]
fn openai_projection_uses_raw_tool_arguments_when_matching_input() {
let msg = ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "call_1".into(),
name: "bash".into(),
input: json!({"b": 2, "a": 1}),
raw_emitted_args: Some(r#"{ "b": 2, "a": 1 }"#.into()),
}],
thinking: None,
usage: None,
};
let wire = chat_message_to_wire(&msg);
assert_eq!(
wire["tool_calls"][0]["function"]["arguments"],
r#"{ "b": 2, "a": 1 }"#
);
}
#[test]
fn openai_projection_ignores_stale_raw_tool_arguments() {
let msg = ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "call_1".into(),
name: "bash".into(),
input: json!({"command": "pwd"}),
raw_emitted_args: Some(r#"{"command": "rm -rf /"}"#.into()),
}],
thinking: None,
usage: None,
};
let wire = chat_message_to_wire(&msg);
assert_eq!(
wire["tool_calls"][0]["function"]["arguments"],
json!({"command": "pwd"}).to_string()
);
}
#[test]
fn map_openai_finish_reason_table() {
assert_eq!(map_openai_finish_reason(Some("stop")), "end_turn");
assert_eq!(map_openai_finish_reason(Some("length")), "max_tokens");
assert_eq!(map_openai_finish_reason(Some("tool_calls")), "end_turn");
assert_eq!(map_openai_finish_reason(Some("content_filter")), "refusal");
assert_eq!(map_openai_finish_reason(None), "end_turn");
assert_eq!(map_openai_finish_reason(Some("")), "end_turn");
}
#[test]
fn chat_message_to_openai_wire_text_only_keeps_string_content() {
let msg = ChatMessage::User {
content: "hello".into(),
attachments: vec![],
};
let v = chat_message_to_wire(&msg);
assert_eq!(v["role"], "user");
assert_eq!(v["content"], "hello");
assert!(v["content"].is_string());
}
#[test]
fn chat_message_to_openai_wire_with_base64_image() {
let msg = ChatMessage::User {
content: "describe this".into(),
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/png".into(),
data: ImageData::Base64("iVBORw0KG...".into()),
})],
};
let v = chat_message_to_wire(&msg);
let parts = v["content"].as_array().expect("content array");
assert_eq!(parts.len(), 2);
assert_eq!(parts[0]["type"], "text");
assert_eq!(parts[0]["text"], "describe this");
assert_eq!(parts[1]["type"], "image_url");
let url = parts[1]["image_url"]["url"].as_str().unwrap();
assert!(url.starts_with("data:image/png;base64,"));
assert!(url.contains("iVBORw0KG..."));
}
#[test]
fn chat_message_to_openai_wire_with_url_image() {
let msg = ChatMessage::User {
content: "".into(), attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/jpeg".into(),
data: ImageData::Url("https://cdn.example.com/cat.jpg".into()),
})],
};
let v = chat_message_to_wire(&msg);
let parts = v["content"].as_array().unwrap();
assert_eq!(parts.len(), 1);
assert_eq!(parts[0]["type"], "image_url");
assert_eq!(
parts[0]["image_url"]["url"],
"https://cdn.example.com/cat.jpg"
);
}
#[test]
fn chat_message_to_openai_tool_role_degrades_image_to_placeholder() {
let msg = ChatMessage::Tool {
tool_call_id: "call_x".into(),
content: "ok".into(),
is_error: false,
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/png".into(),
data: ImageData::Base64("AAA".into()),
})],
};
let v = chat_message_to_wire(&msg);
assert_eq!(v["role"], "tool");
assert_eq!(v["tool_call_id"], "call_x");
let content = v["content"].as_str().unwrap();
assert!(content.starts_with("ok\n"));
assert!(content.contains("image attached: image/png"));
assert!(!content.contains("AAA"));
}
#[test]
fn replay_compaction_leaves_small_results_untouched() {
let small = "x".repeat(1_000);
assert!(matches!(
compact_tool_result_for_replay(&small),
std::borrow::Cow::Borrowed(_)
));
let medium = "word ".repeat(2_000);
let out = compact_tool_result_for_replay(&medium);
assert!(out.contains("compacted for model replay"));
}
#[test]
fn replay_compaction_keeps_head_and_tail_deterministically() {
let body = format!("HEAD_MARK{}TAIL_MARK", "x".repeat(20_000));
let first = compact_tool_result_for_replay(&body).into_owned();
let second = compact_tool_result_for_replay(&body).into_owned();
assert_eq!(first, second);
assert!(first.starts_with("[tool result compacted for model replay]"));
assert!(first.contains("HEAD_MARK"), "head survives");
assert!(first.contains("TAIL_MARK"), "tail survives");
assert!(first.contains("omitted"), "omission marker present");
assert!(first.len() < body.len() / 2);
}
#[test]
fn openai_projection_compacts_oversized_tool_result() {
let big = format!("START{}END", "y".repeat(20_000));
let msg = ChatMessage::Tool {
tool_call_id: "call_big".into(),
content: big.clone(),
is_error: false,
attachments: vec![],
};
let v = chat_message_to_wire(&msg);
let content = v["content"].as_str().unwrap();
assert!(content.contains("compacted for model replay"));
assert!(content.contains("START") && content.contains("END"));
match &msg {
ChatMessage::Tool { content, .. } => assert_eq!(content.len(), big.len()),
_ => unreachable!(),
}
}
#[test]
fn anthropic_projection_compacts_oversized_tool_result() {
let big = "z".repeat(20_000);
let msgs = vec![
ChatMessage::Assistant {
text: None,
tool_calls: vec![crate::tools::ToolInvocation {
id: "tc_big".into(),
name: "bash".into(),
input: json!({}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "tc_big".into(),
content: big,
is_error: false,
attachments: vec![],
},
];
let wire = chat_messages_to_anthropic_messages(&msgs);
let rendered = serde_json::to_string(&wire).unwrap();
assert!(rendered.contains("compacted for model replay"));
}
#[test]
fn chat_messages_to_anthropic_tool_result_carries_image_block() {
let msgs = vec![
ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "tc_img".into(),
name: "screenshot".into(),
input: json!({}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "tc_img".into(),
content: "see image".into(),
is_error: false,
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/png".into(),
data: ImageData::Base64("PNGBYTES".into()),
})],
},
];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 2);
let user = &out[1];
assert_eq!(user["role"], "user");
let outer = user["content"].as_array().unwrap();
assert_eq!(outer.len(), 1);
assert_eq!(outer[0]["type"], "tool_result");
assert_eq!(outer[0]["tool_use_id"], "tc_img");
let inner = outer[0]["content"].as_array().unwrap();
assert_eq!(inner.len(), 2);
assert_eq!(inner[0]["type"], "text");
assert_eq!(inner[0]["text"], "see image");
assert_eq!(inner[1]["type"], "image");
assert_eq!(inner[1]["source"]["type"], "base64");
assert_eq!(inner[1]["source"]["media_type"], "image/png");
assert_eq!(inner[1]["source"]["data"], "PNGBYTES");
}
#[test]
fn chat_messages_to_anthropic_renders_user_text_with_image_block() {
let msgs = vec![ChatMessage::User {
content: "what is this".into(),
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/png".into(),
data: ImageData::Base64("AAAA".into()),
})],
}];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 1);
let blocks = out[0]["content"].as_array().unwrap();
assert_eq!(blocks[0]["type"], "text");
assert_eq!(blocks[0]["text"], "what is this");
assert_eq!(blocks[1]["type"], "image");
assert_eq!(blocks[1]["source"]["type"], "base64");
assert_eq!(blocks[1]["source"]["media_type"], "image/png");
assert_eq!(blocks[1]["source"]["data"], "AAAA");
}
#[test]
fn chat_messages_to_anthropic_renders_url_image() {
let msgs = vec![ChatMessage::User {
content: "".into(),
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/jpeg".into(),
data: ImageData::Url("https://example.com/x.jpg".into()),
})],
}];
let out = chat_messages_to_anthropic_messages(&msgs);
let blocks = out[0]["content"].as_array().unwrap();
assert_eq!(blocks.len(), 1);
assert_eq!(blocks[0]["type"], "image");
assert_eq!(blocks[0]["source"]["type"], "url");
assert_eq!(blocks[0]["source"]["url"], "https://example.com/x.jpg");
}
#[test]
fn chat_message_to_anthropic_merges_tool_results_and_image() {
let msgs = vec![
ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "tc_1".into(),
name: "screenshot".into(),
input: json!({}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "tc_1".into(),
content: "captured".into(),
is_error: false,
attachments: vec![],
},
ChatMessage::User {
content: "what changed?".into(),
attachments: vec![UserAttachment::Image(ImageSource {
media_type: "image/png".into(),
data: ImageData::Base64("ZZ".into()),
})],
},
];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 2);
let blocks = out[1]["content"].as_array().unwrap();
assert_eq!(blocks.len(), 3);
assert_eq!(blocks[0]["type"], "tool_result");
assert_eq!(blocks[0]["tool_use_id"], "tc_1");
assert_eq!(blocks[1]["type"], "text");
assert_eq!(blocks[1]["text"], "what changed?");
assert_eq!(blocks[2]["type"], "image");
}
#[test]
fn chat_messages_to_anthropic_renders_simple_user_assistant() {
let msgs = vec![
ChatMessage::User {
content: "hi".into(),
attachments: vec![],
},
ChatMessage::Assistant {
text: Some("hello".into()),
tool_calls: vec![],
thinking: None,
usage: None,
},
];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 2);
assert_eq!(out[0]["role"], "user");
assert_eq!(out[0]["content"][0]["type"], "text");
assert_eq!(out[0]["content"][0]["text"], "hi");
assert_eq!(out[1]["role"], "assistant");
assert_eq!(out[1]["content"][0]["text"], "hello");
}
#[test]
fn chat_messages_to_anthropic_folds_tool_results_into_next_user() {
let msgs = vec![
ChatMessage::User {
content: "do it".into(),
attachments: vec![],
},
ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "call_1".into(),
name: "bash".into(),
input: json!({"command": "pwd"}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "call_1".into(),
content: "{\"stdout\":\"/\"}".into(),
is_error: false,
attachments: vec![],
},
ChatMessage::User {
content: "explain".into(),
attachments: vec![],
},
];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 3);
assert_eq!(out[1]["role"], "assistant");
assert_eq!(out[1]["content"][0]["type"], "tool_use");
assert_eq!(out[1]["content"][0]["id"], "call_1");
assert_eq!(out[2]["role"], "user");
assert_eq!(out[2]["content"][0]["type"], "tool_result");
assert_eq!(out[2]["content"][0]["tool_use_id"], "call_1");
assert_eq!(out[2]["content"][1]["type"], "text");
assert_eq!(out[2]["content"][1]["text"], "explain");
}
#[test]
fn chat_messages_to_anthropic_renders_thinking_then_text_then_tool_use() {
let msgs = vec![ChatMessage::Assistant {
text: Some("preface".into()),
tool_calls: vec![ToolInvocation {
id: "t".into(),
name: "n".into(),
input: json!({"a": 1}),
raw_emitted_args: None,
}],
thinking: Some(AssistantThinking {
text: "deep thought".into(),
signature: Some("sig123".into()),
}),
usage: None,
}];
let out = chat_messages_to_anthropic_messages(&msgs);
let blocks = out[0]["content"].as_array().unwrap();
assert_eq!(blocks[0]["type"], "thinking");
assert_eq!(blocks[0]["thinking"], "deep thought");
assert_eq!(blocks[0]["signature"], "sig123");
assert_eq!(blocks[1]["type"], "text");
assert_eq!(blocks[1]["text"], "preface");
assert_eq!(blocks[2]["type"], "tool_use");
}
#[test]
fn chat_messages_to_anthropic_trailing_tool_results_flushed() {
let msgs = vec![
ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "t".into(),
name: "n".into(),
input: json!({}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "t".into(),
content: "ok".into(),
is_error: false,
attachments: vec![],
},
];
let out = chat_messages_to_anthropic_messages(&msgs);
assert_eq!(out.len(), 2);
assert_eq!(out[1]["role"], "user");
assert_eq!(out[1]["content"][0]["type"], "tool_result");
}
#[test]
fn apply_anthropic_cache_strategy_marks_system_last_tool_and_last_message() {
let system = anthropic_system_field(Some("system prompt"));
let tools = vec![
json!({"name": "a", "description": "", "input_schema": {"type": "object"}}),
json!({"name": "b", "description": "", "input_schema": {"type": "object"}}),
];
let messages = vec![
json!({"role": "user", "content": [{"type": "text", "text": "hi"}]}),
json!({"role": "assistant", "content": [{"type": "text", "text": "hello"}]}),
];
let out = apply_anthropic_cache_strategy(system, tools, messages);
let sys_block = &out.system.as_ref().unwrap()[0];
assert_eq!(sys_block["cache_control"]["type"], "ephemeral");
assert!(out.tools[0].get("cache_control").is_none());
assert_eq!(out.tools[1]["cache_control"]["type"], "ephemeral");
let last_msg_blocks = out.messages.last().unwrap()["content"].as_array().unwrap();
assert_eq!(
last_msg_blocks.last().unwrap()["cache_control"]["type"],
"ephemeral"
);
}
#[test]
fn apply_anthropic_cache_strategy_skips_empty_system() {
let out = apply_anthropic_cache_strategy(None, vec![], vec![]);
assert!(out.system.is_none());
}
fn anthropic_client_for_tool_choice_tests() -> AnthropicModelClient {
AnthropicModelClient::new(AnthropicConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: resolved_model(
"claude-test",
WireProtocol::Anthropic,
1_024,
None,
ReasoningConfig::default(),
),
anthropic_version: AnthropicConfig::DEFAULT_VERSION.into(),
})
}
#[test]
fn anthropic_client_omits_tool_choice_when_auto() {
let client = anthropic_client_for_tool_choice_tests();
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
});
assert!(body["tools"].as_array().unwrap().len() > 0);
assert!(body.get("tool_choice").is_none());
assert!(body.get("parallel_tool_calls").is_none());
}
#[test]
fn anthropic_client_emits_tool_choice_required_as_any() {
let client = anthropic_client_for_tool_choice_tests();
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Required,
parallel_tool_calls: Some(true),
});
assert_eq!(body["tool_choice"]["type"], "any");
assert!(body.get("parallel_tool_calls").is_none());
}
#[test]
fn anthropic_client_projects_hosted_web_search_tool() {
let client = anthropic_client_for_tool_choice_tests();
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "research current AI market".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![HostedTool::WebSearch],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
});
let tools = body["tools"].as_array().unwrap();
let web = tools
.iter()
.find(|tool| tool.get("name").and_then(Value::as_str) == Some("web_search"))
.unwrap();
assert_eq!(web["type"], "web_search_20250305");
assert!(web.get("max_uses").is_none());
}
#[test]
fn clients_report_web_search_support_by_protocol_and_endpoint() {
let chat = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://api.openai.com/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiCompatible),
});
assert_eq!(
chat.hosted_capability(HostedCapability::WebSearch),
CapabilitySupport::Unsupported
);
let responses = OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: "https://api.openai.com/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiResponses),
reasoning_summary: None,
});
assert_eq!(
responses.hosted_capability(HostedCapability::WebSearch),
CapabilitySupport::Supported
);
let gateway = OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: "https://gateway.example/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-test", WireProtocol::OpenAiResponses),
reasoning_summary: None,
});
assert_eq!(
gateway.hosted_capability(HostedCapability::WebSearch),
CapabilitySupport::Unknown
);
}
#[test]
fn anthropic_client_emits_tool_choice_named_tool() {
let client = anthropic_client_for_tool_choice_tests();
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::Tool("bash".into()),
parallel_tool_calls: None,
});
assert_eq!(body["tool_choice"]["type"], "tool");
assert_eq!(body["tool_choice"]["name"], "bash");
}
#[test]
fn anthropic_client_drops_tools_when_choice_is_none() {
let client = anthropic_client_for_tool_choice_tests();
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![ChatMessage::User {
content: "go".into(),
attachments: vec![],
}],
tools: vec![bash_spec()],
hosted_tools: vec![],
tool_choice: ToolChoice::None,
parallel_tool_calls: None,
});
assert!(body.get("tools").is_none());
assert!(body.get("tool_choice").is_none());
}
#[test]
fn anthropic_stream_state_text_only() {
let mut s = AnthropicStreamState::default();
let _ = s
.feed_event(
"message_start",
r#"{"type":"message_start","message":{"id":"msg_01","usage":{"input_tokens":10,"output_tokens":0}}}"#,
)
.unwrap();
let _ = s
.feed_event(
"content_block_start",
r#"{"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}"#,
)
.unwrap();
let out = s
.feed_event(
"content_block_delta",
r#"{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Hello"}}"#,
)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::TextDelta { msg_id, delta } => {
assert_eq!(msg_id, "msg_01");
assert_eq!(delta, "Hello");
}
other => panic!("expected TextDelta, got {other:?}"),
}
let _ = s.feed_event(
"message_delta",
r#"{"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}"#,
);
let out = s
.feed_event("message_stop", r#"{"type":"message_stop"}"#)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::Done { stop_reason, usage } => {
assert_eq!(stop_reason, "end_turn");
let u = usage.as_ref().unwrap();
assert_eq!(u.input_tokens, 10);
assert_eq!(u.output_tokens, 5);
}
other => panic!("expected Done, got {other:?}"),
}
}
#[test]
fn anthropic_stream_state_thinking_block_emits_delta_and_signature() {
let mut s = AnthropicStreamState::default();
let _ = s.feed_event(
"message_start",
r#"{"type":"message_start","message":{"id":"msg_t"}}"#,
);
let _ = s.feed_event(
"content_block_start",
r#"{"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}}"#,
);
let out = s
.feed_event(
"content_block_delta",
r#"{"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"reasoning..."}}"#,
)
.unwrap();
assert_eq!(out.len(), 1);
let ModelChunk::ThinkingDelta {
delta, signature, ..
} = &out[0]
else {
panic!("expected ThinkingDelta");
};
assert_eq!(delta, "reasoning...");
assert!(signature.is_none());
let out = s
.feed_event(
"content_block_delta",
r#"{"type":"content_block_delta","index":0,"delta":{"type":"signature_delta","signature":"sig_abc"}}"#,
)
.unwrap();
let ModelChunk::ThinkingDelta {
delta, signature, ..
} = &out[0]
else {
panic!("expected ThinkingDelta");
};
assert_eq!(delta, "");
assert_eq!(signature.as_deref(), Some("sig_abc"));
}
#[test]
fn anthropic_stream_state_tool_use_streamed_input() {
let mut s = AnthropicStreamState::default();
let _ = s.feed_event(
"message_start",
r#"{"type":"message_start","message":{"id":"msg_x"}}"#,
);
let out = s
.feed_event(
"content_block_start",
r#"{"type":"content_block_start","index":0,"content_block":{"type":"tool_use","id":"toolu_1","name":"bash","input":{}}}"#,
)
.unwrap();
assert_eq!(out.len(), 1);
match &out[0] {
ModelChunk::ToolCallStart { id, name } => {
assert_eq!(id, "toolu_1");
assert_eq!(name, "bash");
}
other => panic!("expected ToolCallStart, got {other:?}"),
}
let out = s
.feed_event(
"content_block_delta",
r#"{"type":"content_block_delta","index":0,"delta":{"type":"input_json_delta","partial_json":"{\"cmd\":"}}"#,
)
.unwrap();
let ModelChunk::ToolCallInputDelta { id, delta } = &out[0] else {
panic!("expected ToolCallInputDelta");
};
assert_eq!(id, "toolu_1");
assert_eq!(delta, "{\"cmd\":");
let out = s
.feed_event(
"content_block_stop",
r#"{"type":"content_block_stop","index":0}"#,
)
.unwrap();
match &out[0] {
ModelChunk::ToolCallEnd { id, input } => {
assert_eq!(id, "toolu_1");
assert!(input.is_none());
}
other => panic!("expected ToolCallEnd, got {other:?}"),
}
}
#[test]
fn anthropic_stream_state_finalises_on_close_without_message_stop() {
let mut s = AnthropicStreamState::default();
let _ = s
.feed_event(
"message_delta",
r#"{"type":"message_delta","delta":{"stop_reason":"max_tokens"},"usage":{"output_tokens":100}}"#,
)
.unwrap();
let done = s.finalize().unwrap();
match done {
ModelChunk::Done { stop_reason, .. } => assert_eq!(stop_reason, "max_tokens"),
other => panic!("expected Done, got {other:?}"),
}
}
#[test]
fn anthropic_stop_reason_mapping() {
assert_eq!(map_anthropic_stop_reason(Some("end_turn")), "end_turn");
assert_eq!(map_anthropic_stop_reason(Some("tool_use")), "end_turn");
assert_eq!(map_anthropic_stop_reason(Some("max_tokens")), "max_tokens");
assert_eq!(map_anthropic_stop_reason(Some("stop_sequence")), "end_turn");
assert_eq!(map_anthropic_stop_reason(Some("refusal")), "refusal");
assert_eq!(map_anthropic_stop_reason(None), "end_turn");
}
#[test]
fn classify_anthropic_http_error_buckets_by_status() {
use reqwest::StatusCode;
assert!(matches!(
classify_anthropic_http_error(StatusCode::TOO_MANY_REQUESTS, "{}"),
ModelClientError::RateLimit(_)
));
assert!(matches!(
classify_anthropic_http_error(StatusCode::UNAUTHORIZED, "{}"),
ModelClientError::Auth(_)
));
assert!(matches!(
classify_anthropic_http_error(
StatusCode::BAD_REQUEST,
"{\"error\":{\"message\":\"prompt is too long; context_length_exceeded\"}}"
),
ModelClientError::ContextOverflow(_)
));
assert!(matches!(
classify_anthropic_http_error(StatusCode::BAD_REQUEST, "invalid model"),
ModelClientError::BadRequest(_)
));
assert!(matches!(
classify_anthropic_http_error(StatusCode::INTERNAL_SERVER_ERROR, "oops"),
ModelClientError::ServerError(_)
));
}
#[tokio::test]
async fn collect_model_response_folds_streamed_tool_call_arguments() {
let chunks = vec![
Ok(ModelChunk::TextDelta {
msg_id: "m".into(),
delta: "ok ".into(),
}),
Ok(ModelChunk::ToolCallStart {
id: "call_1".into(),
name: "bash".into(),
}),
Ok(ModelChunk::ToolCallInputDelta {
id: "call_1".into(),
delta: "{\"command\":".into(),
}),
Ok(ModelChunk::ToolCallInputDelta {
id: "call_1".into(),
delta: "\"pwd\"}".into(),
}),
Ok(ModelChunk::ToolCallEnd {
id: "call_1".into(),
input: None,
}),
Ok(ModelChunk::Done {
stop_reason: "end_turn".into(),
usage: None,
}),
];
let stream = futures::stream::iter(chunks).boxed();
let response = collect_model_response(stream).await.unwrap();
let ModelResponse::ToolCall {
invocation,
preface,
..
} = response
else {
panic!("expected ToolCall");
};
assert_eq!(invocation.name, "bash");
assert_eq!(invocation.input["command"], "pwd");
assert_eq!(preface.as_deref(), Some("ok "));
}
#[test]
fn classify_openai_http_error_buckets_by_status_and_body() {
use reqwest::StatusCode;
assert!(matches!(
classify_openai_http_error(StatusCode::TOO_MANY_REQUESTS, "rate limit hit"),
ModelClientError::RateLimit(_)
));
assert!(matches!(
classify_openai_http_error(StatusCode::UNAUTHORIZED, "bad key"),
ModelClientError::Auth(_)
));
assert!(matches!(
classify_openai_http_error(StatusCode::FORBIDDEN, "no access"),
ModelClientError::Auth(_)
));
assert!(matches!(
classify_openai_http_error(
StatusCode::BAD_REQUEST,
"{\"error\":{\"message\":\"this model's maximum context length is 8192\"}}"
),
ModelClientError::ContextOverflow(_)
));
assert!(matches!(
classify_openai_http_error(StatusCode::BAD_REQUEST, "missing argument"),
ModelClientError::BadRequest(_)
));
assert!(matches!(
classify_openai_http_error(StatusCode::INTERNAL_SERVER_ERROR, "oops"),
ModelClientError::ServerError(_)
));
}
#[test]
fn looks_like_context_overflow_matches_common_phrasings() {
assert!(looks_like_context_overflow(
"context_length_exceeded: this model has a maximum context length of 8192"
));
assert!(looks_like_context_overflow("too many tokens in prompt"));
assert!(looks_like_context_overflow(
"Prompt exceeds the model's maximum context"
));
assert!(!looks_like_context_overflow("invalid api key"));
}
#[test]
fn parse_openai_usage_returns_none_for_missing_or_all_zero() {
assert!(parse_openai_usage(None).is_none());
assert!(parse_openai_usage(Some(&json!({
"prompt_tokens": 0,
"completion_tokens": 0
})))
.is_none());
}
#[test]
fn openai_client_renders_multi_turn_history() {
let client = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "sk-test".into(),
model: resolved_model(
"gpt-test",
WireProtocol::OpenAiCompatible,
128,
Some(0.2),
ReasoningConfig::default(),
),
});
let body = client.request_body(&ModelTurnInput {
system_prompt: None,
messages: vec![
ChatMessage::User {
content: "run pwd".into(),
attachments: vec![],
},
ChatMessage::Assistant {
text: None,
tool_calls: vec![ToolInvocation {
id: "call_1".into(),
name: "bash".into(),
input: json!({"command": "pwd"}),
raw_emitted_args: None,
}],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: "call_1".into(),
content: "{\"stdout\":\"/home/user\"}".into(),
is_error: false,
attachments: vec![],
},
],
tools: vec![],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
});
assert_eq!(body["temperature"], 0.2);
assert_eq!(body["max_tokens"], 128);
assert_eq!(body["messages"][0]["role"], "user");
assert_eq!(body["messages"][1]["role"], "assistant");
assert_eq!(body["messages"][1]["tool_calls"][0]["id"], "call_1");
assert_eq!(
body["messages"][1]["tool_calls"][0]["function"]["name"],
"bash"
);
assert_eq!(body["messages"][2]["role"], "tool");
assert_eq!(body["messages"][2]["tool_call_id"], "call_1");
}
#[test]
fn reasoning_controls_translate_by_wire_protocol() {
let openai = OpenAiCompatibleModelClient::new(OpenAiCompatibleConfig {
base_url: "https://example.test".into(),
api_key: "test".into(),
model: resolved_model(
"deepseek-v4-pro",
WireProtocol::OpenAiCompatible,
4_096,
None,
ReasoningConfig {
mode: ReasoningMode::Enabled,
effort: None,
budget_tokens: None,
},
),
});
let body = openai.request_body(&user("hi"));
assert_eq!(body["thinking"]["type"], "enabled");
let anthropic = AnthropicModelClient::new(AnthropicConfig {
base_url: "https://example.test/v1".into(),
api_key: "test".into(),
model: resolved_model(
"claude-sonnet-4-6",
WireProtocol::Anthropic,
4_096,
None,
ReasoningConfig {
mode: ReasoningMode::Enabled,
effort: Some("high".into()),
budget_tokens: None,
},
),
anthropic_version: AnthropicConfig::DEFAULT_VERSION.into(),
});
let body = anthropic.request_body(&user("hi"));
assert_eq!(body["thinking"]["type"], "adaptive");
assert_eq!(body["output_config"]["effort"], "high");
let responses = OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: "https://example.test/v1".into(),
api_key: "test".into(),
model: resolved_model(
"gpt-5.5",
WireProtocol::OpenAiResponses,
4_096,
None,
ReasoningConfig {
mode: ReasoningMode::Disabled,
effort: Some("none".into()),
budget_tokens: None,
},
),
reasoning_summary: None,
});
let body = responses.request_body(&user("hi"));
assert_eq!(body["reasoning"]["effort"], "none");
}
#[tokio::test]
async fn scripted_client_emits_tool_call_then_summary() {
let scripted = ScriptedModelClient;
let first = scripted
.next(user("read README.md"))
.await
.expect("scripted first");
let ModelResponse::ToolCall { invocation, .. } = first else {
panic!("expected tool call on first step");
};
assert_eq!(invocation.name, "read");
let history = ModelTurnInput {
system_prompt: None,
messages: vec![
ChatMessage::User {
content: "read README.md".into(),
attachments: vec![],
},
ChatMessage::Assistant {
text: None,
tool_calls: vec![invocation.clone()],
thinking: None,
usage: None,
},
ChatMessage::Tool {
tool_call_id: invocation.id.clone(),
content: "{\"content\":\"hi\"}".into(),
is_error: false,
attachments: vec![],
},
],
tools: vec![],
hosted_tools: vec![],
tool_choice: ToolChoice::Auto,
parallel_tool_calls: None,
};
let second = scripted.next(history).await.expect("scripted second");
let ModelResponse::Message { text, .. } = second else {
panic!("expected final message after tool result");
};
assert!(text.contains("completed"));
}
#[test]
fn responses_client_builds_responses_endpoint_and_body() {
let mk = |base: &str| {
OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: base.into(),
api_key: "sk-test".into(),
model: resolved_model(
"gpt-5",
WireProtocol::OpenAiResponses,
2_048,
None,
ReasoningConfig {
mode: ReasoningMode::Enabled,
effort: Some("high".into()),
budget_tokens: None,
},
),
reasoning_summary: None,
})
};
assert_eq!(
mk("https://api.openai.com/v1").endpoint(),
"https://api.openai.com/v1/responses"
);
assert_eq!(
mk("https://api.openai.com/v1/").endpoint(),
"https://api.openai.com/v1/responses"
);
assert_eq!(
mk("https://api.openai.com/v1/responses").endpoint(),
"https://api.openai.com/v1/responses"
);
let client = mk("https://api.openai.com/v1");
let mut input = user("hello");
input.system_prompt = Some("be terse".into());
input.tools = vec![bash_spec()];
let body = client.request_body(&input);
assert_eq!(body["model"], "gpt-5");
assert_eq!(body["stream"], true);
assert_eq!(body["store"], false);
assert_eq!(body["instructions"], "be terse");
assert_eq!(body["max_output_tokens"], 2048);
assert_eq!(body["reasoning"]["effort"], "high");
assert_eq!(body["include"][0], "reasoning.encrypted_content");
assert_eq!(body["tools"][0]["type"], "function");
assert_eq!(body["tools"][0]["name"], "bash");
assert!(body["tools"][0]["parameters"].is_object());
assert_eq!(body["tool_choice"], "auto");
assert_eq!(body["input"][0]["role"], "user");
assert_eq!(body["input"][0]["content"][0]["type"], "input_text");
assert_eq!(body["input"][0]["content"][0]["text"], "hello");
}
#[test]
fn responses_tool_choice_none_drops_client_and_hosted_tools() {
let client = OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: "https://api.openai.com/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-5", WireProtocol::OpenAiResponses),
reasoning_summary: None,
});
let mut input = user("hi");
input.tools = vec![bash_spec()];
input.hosted_tools = vec![HostedTool::WebSearch];
input.tool_choice = ToolChoice::None;
let body = client.request_body(&input);
assert!(
body.get("tools").is_none(),
"tools should be absent, got {:?}",
body.get("tools")
);
assert!(body.get("tool_choice").is_none());
input.tool_choice = ToolChoice::Auto;
let body = client.request_body(&input);
let tools = body["tools"].as_array().expect("tools present");
assert_eq!(tools.len(), 2);
assert!(tools.iter().any(|t| t["type"] == "function"));
assert!(tools.iter().any(|t| t["type"] == "web_search"));
}
#[test]
fn responses_tool_choice_required_sent_for_hosted_only() {
let client = OpenAiResponsesModelClient::new(OpenAiResponsesConfig {
base_url: "https://api.openai.com/v1".into(),
api_key: "sk-test".into(),
model: default_model("gpt-5", WireProtocol::OpenAiResponses),
reasoning_summary: None,
});
let mut input = user("search the web");
input.hosted_tools = vec![HostedTool::WebSearch];
input.tool_choice = ToolChoice::Required;
let body = client.request_body(&input);
assert_eq!(body["tools"][0]["type"], "web_search");
assert_eq!(body["tool_choice"], "required");
input.tool_choice = ToolChoice::Tool("bash".into());
let body = client.request_body(&input);
assert_eq!(body["tools"][0]["type"], "web_search");
assert!(
body.get("tool_choice").is_none(),
"Tool(name) must not be sent for a hosted-only turn, got {:?}",
body.get("tool_choice")
);
input.tools = vec![bash_spec()];
let body = client.request_body(&input);
assert_eq!(body["tool_choice"]["type"], "function");
assert_eq!(body["tool_choice"]["name"], "bash");
}
#[test]
fn responses_stream_state_accepts_reasoning_summary_alias() {
let mut st = OpenAiResponsesStreamState::default();
let chunks = st
.feed_data(
r#"{"type":"response.reasoning_summary.delta","item_id":"rs_1","summary_index":0,"delta":"aliased"}"#,
)
.expect("feed_data");
assert!(
matches!(&chunks[0], ModelChunk::ThinkingDelta { delta, .. } if delta == "aliased")
);
}
#[test]
fn responses_stream_state_separates_reasoning_summary_parts() {
let mut st = OpenAiResponsesStreamState::default();
let mut chunks: Vec<ModelChunk> = Vec::new();
let mut feed = |st: &mut OpenAiResponsesStreamState, data: &str| {
chunks.extend(st.feed_data(data).expect("feed_data"));
};
feed(
&mut st,
r#"{"type":"response.reasoning_summary_part.added","item_id":"rs_1","summary_index":0}"#,
);
feed(
&mut st,
r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","summary_index":0,"delta":"first"}"#,
);
feed(
&mut st,
r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","summary_index":0,"delta":" more"}"#,
);
feed(
&mut st,
r#"{"type":"response.reasoning_summary_part.added","item_id":"rs_1","summary_index":1}"#,
);
feed(
&mut st,
r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","summary_index":1,"delta":"second"}"#,
);
let deltas: Vec<&str> = chunks
.iter()
.filter_map(|c| match c {
ModelChunk::ThinkingDelta { delta, .. } => Some(delta.as_str()),
_ => None,
})
.collect();
assert_eq!(deltas, vec!["first", " more", "\n\nsecond"]);
}
#[test]
fn responses_input_projection_round_trips_tool_calls_and_reasoning() {
let sig = encode_reasoning_signature("rs_123", "ENC==");
let messages = vec![
ChatMessage::User {
content: "hi".into(),
attachments: vec![],
},
ChatMessage::Assistant {
text: Some("looked".into()),
tool_calls: vec![ToolInvocation {
id: "call_1".into(),
name: "bash".into(),
input: json!({"command": "ls"}),
raw_emitted_args: None,
}],
thinking: Some(AssistantThinking {
text: "let me look".into(),
signature: Some(sig),
}),
usage: None,
},
ChatMessage::Tool {
tool_call_id: "call_1".into(),
content: "file.txt".into(),
is_error: false,
attachments: vec![],
},
];
let input = chat_messages_to_responses_input(&messages);
assert_eq!(input[0]["role"], "user");
assert_eq!(input[1]["type"], "reasoning");
assert_eq!(input[1]["id"], "rs_123");
assert_eq!(input[1]["encrypted_content"], "ENC==");
assert_eq!(input[1]["summary"][0]["text"], "let me look");
assert_eq!(input[2]["role"], "assistant");
assert_eq!(input[2]["content"][0]["type"], "output_text");
assert_eq!(input[3]["type"], "function_call");
assert_eq!(input[3]["call_id"], "call_1");
assert_eq!(input[3]["name"], "bash");
assert_eq!(input[4]["type"], "function_call_output");
assert_eq!(input[4]["call_id"], "call_1");
assert_eq!(input[4]["output"], "file.txt");
}
#[test]
fn responses_stream_state_emits_text_toolcall_and_done() {
let mut st = OpenAiResponsesStreamState::default();
let mut chunks: Vec<ModelChunk> = Vec::new();
let mut feed = |st: &mut OpenAiResponsesStreamState, data: &str| {
chunks.extend(st.feed_data(data).expect("feed_data"));
};
feed(
&mut st,
r#"{"type":"response.created","response":{"id":"resp_1"}}"#,
);
feed(
&mut st,
r#"{"type":"response.output_text.delta","item_id":"msg_1","delta":"Hel"}"#,
);
feed(
&mut st,
r#"{"type":"response.output_text.delta","item_id":"msg_1","delta":"lo"}"#,
);
feed(
&mut st,
r#"{"type":"response.output_item.added","item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"bash"}}"#,
);
feed(
&mut st,
r#"{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"{\"command\":"}"#,
);
feed(
&mut st,
r#"{"type":"response.function_call_arguments.delta","item_id":"fc_1","delta":"\"ls\"}"}"#,
);
feed(
&mut st,
r#"{"type":"response.output_item.done","item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"bash","arguments":"{\"command\":\"ls\"}"}}"#,
);
feed(
&mut st,
r#"{"type":"response.completed","response":{"id":"resp_1","usage":{"input_tokens":10,"output_tokens":5,"input_tokens_details":{"cached_tokens":2}}}}"#,
);
assert!(matches!(&chunks[0], ModelChunk::TextDelta { delta, .. } if delta == "Hel"));
assert!(matches!(&chunks[1], ModelChunk::TextDelta { delta, .. } if delta == "lo"));
assert!(
matches!(&chunks[2], ModelChunk::ToolCallStart { id, name } if id == "call_1" && name == "bash")
);
assert!(matches!(&chunks[3], ModelChunk::ToolCallInputDelta { id, .. } if id == "call_1"));
assert!(matches!(&chunks[4], ModelChunk::ToolCallInputDelta { id, .. } if id == "call_1"));
match &chunks[5] {
ModelChunk::ToolCallEnd { id, input } => {
assert_eq!(id, "call_1");
assert_eq!(input.as_ref().expect("early input")["command"], "ls");
}
other => panic!("expected ToolCallEnd, got {other:?}"),
}
match chunks.last().expect("done chunk") {
ModelChunk::Done { stop_reason, usage } => {
assert_eq!(stop_reason, "end_turn");
let u = usage.as_ref().expect("usage");
assert_eq!(u.input_tokens, 10);
assert_eq!(u.output_tokens, 5);
assert_eq!(u.cache_read_input_tokens, 2);
}
other => panic!("expected Done, got {other:?}"),
}
assert!(st.ended_cleanly());
}
#[test]
fn responses_stream_state_round_trips_reasoning_signature() {
let mut st = OpenAiResponsesStreamState::default();
let mut chunks: Vec<ModelChunk> = Vec::new();
chunks.extend(
st.feed_data(
r#"{"type":"response.reasoning_summary_text.delta","item_id":"rs_1","delta":"pondering"}"#,
)
.unwrap(),
);
chunks.extend(
st.feed_data(
r#"{"type":"response.output_item.done","item":{"type":"reasoning","id":"rs_1","encrypted_content":"ENC=="}}"#,
)
.unwrap(),
);
assert!(
matches!(&chunks[0], ModelChunk::ThinkingDelta { delta, signature, .. } if delta == "pondering" && signature.is_none())
);
match &chunks[1] {
ModelChunk::ThinkingDelta {
delta, signature, ..
} => {
assert!(delta.is_empty());
let (id, enc) =
decode_reasoning_signature(signature.as_ref().expect("signature")).unwrap();
assert_eq!(id, "rs_1");
assert_eq!(enc, "ENC==");
}
other => panic!("expected ThinkingDelta, got {other:?}"),
}
}
#[test]
fn responses_stream_state_reports_cutoff_without_terminal_event() {
let mut st = OpenAiResponsesStreamState::default();
st.feed_data(r#"{"type":"response.output_text.delta","item_id":"m","delta":"hi"}"#)
.unwrap();
assert!(!st.ended_cleanly());
}
#[test]
fn reasoning_signature_encode_decode_roundtrip() {
let sig = encode_reasoning_signature("rs_abc", "base64==payload");
assert_eq!(
decode_reasoning_signature(&sig),
Some(("rs_abc".into(), "base64==payload".into()))
);
assert_eq!(decode_reasoning_signature("anthropic-sig-no-newline"), None);
}
}