use std::collections::BTreeMap;
use std::sync::Arc;
use async_trait::async_trait;
use serde_json::{json, Map, Value};
use crate::event_stream::{
create_assistant_message_event_stream, AssistantMessageEventStream,
AssistantMessageEventStreamProducer,
};
use crate::model::{MaxTokensField, Model, StreamingProtocolCompat};
use crate::provider::{Provider, SimpleStreamOptions};
use crate::providers::anthropic::cost::calculate_cost;
use crate::providers::anthropic::json_parse::parse_streaming_json;
use crate::providers::anthropic::retry::retry_provider_request;
use crate::providers::anthropic::sse::SseEventStream;
use crate::types::{
Api, AssistantMessage, AssistantMessageEvent, Content, Context, DoneReason, ErrorReason,
InputModality, Message, StopReason, ThinkingContent, ThinkingContentType, ThinkingLevel,
ToolCall, ToolCallType, Usage, UserContent,
};
use crate::AiError;
pub struct OpenAiCompletionsProvider {
id: String,
api_key: Option<String>,
http: reqwest::Client,
models: Vec<Model>,
}
impl OpenAiCompletionsProvider {
pub fn with_models(
id: impl Into<String>,
api_key: Option<String>,
http: reqwest::Client,
models: Vec<Model>,
) -> Self {
Self {
id: id.into(),
api_key,
http,
models,
}
}
}
#[async_trait]
impl Provider for OpenAiCompletionsProvider {
fn id(&self) -> &str {
&self.id
}
fn models(&self) -> &[Model] {
&self.models
}
async fn stream_simple(
&self,
model: &Model,
ctx: &Context,
opts: &SimpleStreamOptions,
) -> AssistantMessageEventStream {
let (mut producer, stream) = create_assistant_message_event_stream();
let http = self.http.clone();
let provider_key = self.api_key.clone();
let model = model.clone();
let ctx = Arc::new(ctx.clone());
let opts = opts.clone();
tokio::spawn(async move {
run_stream(&mut producer, http, provider_key, &model, &ctx, &opts).await;
});
stream
}
}
async fn run_stream(
producer: &mut AssistantMessageEventStreamProducer,
http: reqwest::Client,
provider_key: Option<String>,
model: &Model,
ctx: &Context,
opts: &SimpleStreamOptions,
) {
let mut state = StreamState::new(model);
let api_key = opts.api_key.clone().or(provider_key).or_else(|| {
std::env::var("OPENAI_API_KEY")
.ok()
.filter(|v| !v.is_empty())
});
let mut headers = model.headers.clone().unwrap_or_default();
if let Some(extra) = &opts.headers {
for (name, value) in extra {
headers.insert(name.clone(), value.clone());
}
}
if let Some(key) = api_key {
headers.retain(|name, _| !name.eq_ignore_ascii_case("authorization"));
headers.insert("authorization".into(), format!("Bearer {key}"));
}
if !has_auth_header(&headers) {
state.error(
producer,
format!("No API key for provider: {}", model.provider),
false,
);
return;
}
let body = build_request(model, ctx, opts);
let url = chat_completions_url(&model.base_url);
let timeout = opts.timeout.or(Some(std::time::Duration::from_secs(120)));
let signal = opts.signal.clone();
let response = retry_provider_request(
move || {
let http = http.clone();
let url = url.clone();
let body = body.clone();
let headers = headers.clone();
let signal = signal.clone();
async move {
let mut request = http.post(&url);
if let Some(timeout) = timeout {
request = request.timeout(timeout);
}
for (name, value) in headers {
request = request.header(name, value);
}
let response = tokio::select! {
result = request.json(&body).send() => result.map_err(|error| AiError::Http {
status: None,
message: format!("http transport error: {error}"),
})?,
_ = signal.cancelled() => return Err(AiError::Abort {
message: "Request aborted".into(),
}),
};
let status = response.status();
if status.is_success() {
Ok(response)
} else {
let code = status.as_u16();
let message = response.text().await.unwrap_or_default();
Err(AiError::Http {
status: Some(code),
message,
})
}
}
},
opts.max_retries,
opts.max_retry_delay,
&opts.signal,
)
.await;
let response = match response {
Ok(response) => response,
Err(error) => {
state.error(producer, error.to_string(), error.is_abort());
return;
}
};
state.start(producer);
let mut events = SseEventStream::new(response, opts.signal.clone());
loop {
match events.next_event().await {
Ok(Some(event)) if event.data.trim() == "[DONE]" => {
state.finish(producer, model);
return;
}
Ok(Some(event)) => match serde_json::from_str::<Value>(&event.data) {
Ok(chunk) => {
if let Some(message) = chunk
.get("error")
.and_then(|v| v.get("message"))
.and_then(Value::as_str)
{
state.error(producer, message.to_string(), false);
return;
}
state.apply_chunk(producer, &chunk);
}
Err(error) => {
state.error(
producer,
format!("failed to parse OpenAI SSE chunk: {error}"),
false,
);
return;
}
},
Ok(None) => {
state.finish(producer, model);
return;
}
Err(error) => {
state.error(producer, error.to_string(), error.is_abort());
return;
}
}
}
}
fn build_request(model: &Model, ctx: &Context, opts: &SimpleStreamOptions) -> Value {
let mut messages = Vec::new();
if let Some(system) = ctx.system_prompt.as_deref().filter(|s| !s.is_empty()) {
let role = if model.reasoning && supports_developer_role(model) {
"developer"
} else {
"system"
};
messages.push(json!({ "role": role, "content": system }));
}
let mut pending_tool_images = Vec::new();
let mut last_was_tool_result = false;
for message in &ctx.messages {
if !matches!(message, Message::ToolResult(_)) {
flush_tool_result_images(model, &mut messages, &mut pending_tool_images);
}
match message {
Message::User(message) => {
if last_was_tool_result && requires_assistant_after_tool_result(model) {
messages.push(json!({
"role": "assistant",
"content": "I have processed the tool results.",
}));
}
messages.push(json!({
"role": "user",
"content": user_content(&message.content),
}));
last_was_tool_result = false;
}
Message::Assistant(message) => {
let assistant_content = if requires_thinking_as_text(model) {
let parts: Vec<Value> = message
.content
.iter()
.filter_map(|content| match content {
Content::Thinking(thinking) if !thinking.thinking.is_empty() => {
Some(json!({ "type": "text", "text": thinking.thinking }))
}
Content::Text(text) if !text.text.is_empty() => {
Some(json!({ "type": "text", "text": text.text }))
}
_ => None,
})
.collect();
if parts.is_empty() {
Value::Null
} else {
Value::Array(parts)
}
} else {
let text = Content::text_only(&message.content, "\n");
if text.is_empty() {
Value::Null
} else {
Value::String(text)
}
};
let tool_calls: Vec<Value> = message
.content
.iter()
.filter_map(|content| match content {
Content::ToolCall(call) => Some(json!({
"id": call.id,
"type": "function",
"function": {
"name": call.name,
"arguments": call.arguments.to_string(),
}
})),
_ => None,
})
.collect();
let mut value = json!({
"role": "assistant",
"content": assistant_content,
});
if !tool_calls.is_empty() {
value["tool_calls"] = Value::Array(tool_calls);
}
messages.push(value);
last_was_tool_result = false;
}
Message::ToolResult(message) => {
let text = Content::text_only(&message.content, "\n");
let has_images = message
.content
.iter()
.any(|content| matches!(content, Content::Image(_)));
let content = if !text.is_empty() {
text
} else if has_images {
"(see attached image)".to_string()
} else {
"(no tool output)".to_string()
};
let mut value = json!({
"role": "tool",
"tool_call_id": message.tool_call_id,
"content": content,
});
if requires_tool_result_name(model) && !message.tool_name.is_empty() {
value["name"] = Value::String(message.tool_name.clone());
}
messages.push(value);
if model.input.contains(&InputModality::Image) {
pending_tool_images.extend(message.content.iter().filter_map(|content| {
match content {
Content::Image(image) => Some(json!({
"type": "image_url",
"image_url": {
"url": format!("data:{};base64,{}", image.mime_type, image.data)
}
})),
_ => None,
}
}));
}
last_was_tool_result = true;
}
}
}
flush_tool_result_images(model, &mut messages, &mut pending_tool_images);
let mut body = Map::new();
body.insert("model".into(), Value::String(model.id.clone()));
body.insert("messages".into(), Value::Array(messages));
body.insert("stream".into(), Value::Bool(true));
if supports_streaming_usage(model) {
body.insert("stream_options".into(), json!({ "include_usage": true }));
}
let max_tokens = opts.max_tokens.unwrap_or(model.max_tokens);
if max_tokens > 0 {
body.insert(max_tokens_field(model).into(), Value::from(max_tokens));
}
if let Some(temperature) = opts.temperature {
body.insert("temperature".into(), Value::from(temperature));
}
if model.reasoning && supports_reasoning_effort(model) {
if let Some(effort) = reasoning_effort(opts.reasoning_level()) {
body.insert("reasoning_effort".into(), Value::String(effort.into()));
}
}
if !ctx.tools.is_empty() {
body.insert(
"tools".into(),
Value::Array(
ctx.tools
.iter()
.map(|tool| {
json!({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters.as_value(),
}
})
})
.collect(),
),
);
}
if supports_store(model) {
body.insert("store".into(), Value::Bool(false));
}
if let Some(params) = &model.sampling_params {
for (name, value) in params {
body.insert(name.clone(), value.clone());
}
}
Value::Object(body)
}
fn flush_tool_result_images(
model: &Model,
messages: &mut Vec<Value>,
pending_images: &mut Vec<Value>,
) {
if pending_images.is_empty() {
return;
}
if requires_assistant_after_tool_result(model) {
messages.push(json!({
"role": "assistant",
"content": "I have processed the tool results.",
}));
}
let mut content = vec![json!({
"type": "text",
"text": "Attached image(s) from tool result:",
})];
content.append(pending_images);
messages.push(json!({ "role": "user", "content": content }));
}
fn user_content(content: &UserContent) -> Value {
match content {
UserContent::Text(text) => Value::String(text.clone()),
UserContent::Blocks(blocks) => Value::Array(
blocks
.iter()
.filter_map(|block| match block {
Content::Text(text) => Some(json!({ "type": "text", "text": text.text })),
Content::Image(image) => Some(json!({
"type": "image_url",
"image_url": {
"url": format!("data:{};base64,{}", image.mime_type, image.data)
}
})),
_ => None,
})
.collect(),
),
}
}
fn chat_completions_url(base_url: &str) -> String {
let base = base_url.trim_end_matches('/');
if base.ends_with("/v1") {
format!("{base}/chat/completions")
} else {
format!("{base}/v1/chat/completions")
}
}
fn compat(model: &Model) -> Option<&crate::model::OpenaiCompletionsCompat> {
model.compat.as_ref().and_then(|compat| match compat {
StreamingProtocolCompat::OpenaiCompletions(compat) => Some(compat),
_ => None,
})
}
fn supports_streaming_usage(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.supports_usage_in_streaming)
.unwrap_or(true)
}
fn supports_reasoning_effort(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.supports_reasoning_effort)
.unwrap_or(model.reasoning)
}
fn supports_store(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.supports_store)
.unwrap_or(false)
}
fn supports_developer_role(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.supports_developer_role)
.unwrap_or(false)
}
fn supports_finish_reason(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.supports_finish_reason)
.unwrap_or(true)
}
fn requires_tool_result_name(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.requires_tool_result_name)
.unwrap_or(false)
}
fn requires_assistant_after_tool_result(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.requires_assistant_after_tool_result)
.unwrap_or(false)
}
fn requires_thinking_as_text(model: &Model) -> bool {
compat(model)
.and_then(|compat| compat.requires_thinking_as_text)
.unwrap_or(false)
}
fn max_tokens_field(model: &Model) -> &'static str {
match compat(model).and_then(|compat| compat.max_tokens_field) {
Some(MaxTokensField::MaxCompletionTokens) => "max_completion_tokens",
Some(MaxTokensField::MaxTokens) => "max_tokens",
None if model.reasoning => "max_completion_tokens",
None => "max_tokens",
}
}
fn reasoning_effort(level: ThinkingLevel) -> Option<&'static str> {
match level {
ThinkingLevel::Off => None,
ThinkingLevel::Minimal => Some("minimal"),
ThinkingLevel::Low => Some("low"),
ThinkingLevel::Medium => Some("medium"),
ThinkingLevel::High | ThinkingLevel::Xhigh | ThinkingLevel::Max => Some("high"),
}
}
fn has_auth_header(headers: &BTreeMap<String, String>) -> bool {
headers.keys().any(|name| {
name.eq_ignore_ascii_case("authorization") || name.eq_ignore_ascii_case("x-api-key")
})
}
struct ToolState {
content_index: usize,
id: String,
name: String,
arguments: String,
}
struct StreamState {
output: AssistantMessage,
text_index: Option<usize>,
thinking_index: Option<usize>,
tools: BTreeMap<usize, ToolState>,
finish_reason: Option<DoneReason>,
finish_error: Option<String>,
started: bool,
}
impl StreamState {
fn new(model: &Model) -> Self {
Self {
output: AssistantMessage::empty(
Api::OpenaiCompletions,
model.provider.clone(),
model.id.clone(),
now_ms(),
),
text_index: None,
thinking_index: None,
tools: BTreeMap::new(),
finish_reason: None,
finish_error: None,
started: false,
}
}
fn start(&mut self, producer: &mut AssistantMessageEventStreamProducer) {
if !self.started {
self.started = true;
producer.push(AssistantMessageEvent::Start {
partial: Arc::new(self.output.clone()),
});
}
}
fn apply_chunk(&mut self, producer: &mut AssistantMessageEventStreamProducer, chunk: &Value) {
if let Some(id) = chunk.get("id").and_then(Value::as_str) {
self.output.response_id = Some(id.to_string());
}
if let Some(model) = chunk.get("model").and_then(Value::as_str) {
self.output.response_model = Some(model.to_string());
}
if let Some(usage) = chunk.get("usage").filter(|v| !v.is_null()) {
self.output.usage = parse_usage(usage);
}
let Some(choice) = chunk
.get("choices")
.and_then(Value::as_array)
.and_then(|choices| choices.first())
else {
return;
};
if chunk.get("usage").is_none_or(Value::is_null) {
if let Some(usage) = choice.get("usage").filter(|v| !v.is_null()) {
self.output.usage = parse_usage(usage);
}
}
let delta = choice.get("delta").unwrap_or(&Value::Null);
if let Some(text) = delta
.get("content")
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
{
self.push_text(producer, text);
}
let thinking = delta
.get("reasoning_content")
.or_else(|| delta.get("reasoning"))
.or_else(|| delta.get("reasoning_text"))
.and_then(Value::as_str)
.filter(|s| !s.is_empty());
if let Some(thinking) = thinking {
self.push_thinking(producer, thinking);
}
if let Some(tool_calls) = delta.get("tool_calls").and_then(Value::as_array) {
for tool in tool_calls {
self.push_tool_delta(producer, tool);
}
}
if let Some(reason) = choice.get("finish_reason").and_then(Value::as_str) {
self.output.raw_stop_reason = Some(reason.to_string());
match reason {
"stop" | "end" => self.finish_reason = Some(DoneReason::Stop),
"length" => self.finish_reason = Some(DoneReason::Length),
"tool_calls" | "function_call" => self.finish_reason = Some(DoneReason::ToolUse),
other => {
self.finish_error = Some(format!("Provider finish_reason: {other}"));
}
}
}
}
fn push_text(&mut self, producer: &mut AssistantMessageEventStreamProducer, delta: &str) {
let index = match self.text_index {
Some(index) => index,
None => {
let index = self.output.content.len();
self.output.content.push(Content::text(""));
self.text_index = Some(index);
producer.push(AssistantMessageEvent::TextStart {
content_index: index,
partial: Arc::new(self.output.clone()),
});
index
}
};
if let Some(Content::Text(text)) = self.output.content.get_mut(index) {
text.text.push_str(delta);
}
producer.push(AssistantMessageEvent::TextDelta {
content_index: index,
delta: delta.to_string(),
partial: Arc::new(self.output.clone()),
});
}
fn push_thinking(&mut self, producer: &mut AssistantMessageEventStreamProducer, delta: &str) {
let index = match self.thinking_index {
Some(index) => index,
None => {
let index = self.output.content.len();
self.output.content.push(Content::Thinking(ThinkingContent {
kind: ThinkingContentType,
thinking: String::new(),
thinking_signature: None,
redacted: false,
}));
self.thinking_index = Some(index);
producer.push(AssistantMessageEvent::ThinkingStart {
content_index: index,
partial: Arc::new(self.output.clone()),
});
index
}
};
if let Some(Content::Thinking(thinking)) = self.output.content.get_mut(index) {
thinking.thinking.push_str(delta);
}
producer.push(AssistantMessageEvent::ThinkingDelta {
content_index: index,
delta: delta.to_string(),
partial: Arc::new(self.output.clone()),
});
}
fn push_tool_delta(
&mut self,
producer: &mut AssistantMessageEventStreamProducer,
value: &Value,
) {
let wire_index = value.get("index").and_then(Value::as_u64).unwrap_or(0) as usize;
if !self.tools.contains_key(&wire_index) {
let content_index = self.output.content.len();
self.output
.content
.push(Content::tool_call("", "", json!({})));
self.tools.insert(
wire_index,
ToolState {
content_index,
id: String::new(),
name: String::new(),
arguments: String::new(),
},
);
producer.push(AssistantMessageEvent::ToolCallStart {
content_index,
partial: Arc::new(self.output.clone()),
});
}
let tool = self
.tools
.get_mut(&wire_index)
.expect("tool state inserted");
if let Some(id) = value
.get("id")
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
{
tool.id.push_str(id);
}
if let Some(name) = value
.get("function")
.and_then(|v| v.get("name"))
.and_then(Value::as_str)
.filter(|s| !s.is_empty())
{
tool.name.push_str(name);
}
let arguments = value
.get("function")
.and_then(|v| v.get("arguments"))
.and_then(Value::as_str)
.unwrap_or_default();
tool.arguments.push_str(arguments);
if let Some(Content::ToolCall(call)) = self.output.content.get_mut(tool.content_index) {
call.id = tool.id.clone();
call.name = tool.name.clone();
call.arguments = parse_streaming_json(Some(&tool.arguments));
}
if !arguments.is_empty() {
producer.push(AssistantMessageEvent::ToolCallDelta {
content_index: tool.content_index,
delta: arguments.to_string(),
partial: Arc::new(self.output.clone()),
});
}
}
fn finish(&mut self, producer: &mut AssistantMessageEventStreamProducer, model: &Model) {
if let Some(index) = self.text_index {
let content = match self.output.content.get(index) {
Some(Content::Text(text)) => text.text.clone(),
_ => String::new(),
};
producer.push(AssistantMessageEvent::TextEnd {
content_index: index,
content,
partial: Arc::new(self.output.clone()),
});
}
if let Some(index) = self.thinking_index {
let content = match self.output.content.get(index) {
Some(Content::Thinking(thinking)) => thinking.thinking.clone(),
_ => String::new(),
};
producer.push(AssistantMessageEvent::ThinkingEnd {
content_index: index,
content,
partial: Arc::new(self.output.clone()),
});
}
for tool in self.tools.values() {
let call = ToolCall {
kind: ToolCallType,
id: tool.id.clone(),
name: tool.name.clone(),
arguments: parse_streaming_json(Some(&tool.arguments)),
thought_signature: None,
namespace: None,
};
if let Some(slot) = self.output.content.get_mut(tool.content_index) {
*slot = Content::ToolCall(call.clone());
}
producer.push(AssistantMessageEvent::ToolCallEnd {
content_index: tool.content_index,
tool_call: call,
partial: Arc::new(self.output.clone()),
});
}
if let Some(message) = self.finish_error.take() {
self.error(producer, message, false);
return;
}
if self.finish_reason.is_none() && supports_finish_reason(model) {
self.error(
producer,
"Stream ended without finish_reason".to_string(),
false,
);
return;
}
let reason = self.finish_reason.unwrap_or_else(|| {
if self.tools.is_empty() {
DoneReason::Stop
} else {
DoneReason::ToolUse
}
});
self.output.stop_reason = reason.into();
self.output.usage.cost = calculate_cost(&model.cost, &self.output.usage);
producer.push(AssistantMessageEvent::Done {
reason,
message: self.output.clone(),
});
}
fn error(
&mut self,
producer: &mut AssistantMessageEventStreamProducer,
message: String,
aborted: bool,
) {
self.output.stop_reason = if aborted {
StopReason::Aborted
} else {
StopReason::Error
};
self.output.error_message = Some(message);
producer.push(AssistantMessageEvent::Error {
reason: if aborted {
ErrorReason::Aborted
} else {
ErrorReason::Error
},
error: self.output.clone(),
});
}
}
fn parse_usage(value: &Value) -> Usage {
let prompt_tokens = value
.get("prompt_tokens")
.and_then(Value::as_i64)
.unwrap_or(0);
let output = value
.get("completion_tokens")
.and_then(Value::as_i64)
.unwrap_or(0);
let cache_read = value
.get("prompt_tokens_details")
.and_then(|v| v.get("cached_tokens"))
.and_then(Value::as_i64)
.or_else(|| value.get("prompt_cache_hit_tokens").and_then(Value::as_i64))
.or_else(|| value.get("cached_tokens").and_then(Value::as_i64))
.unwrap_or(0);
let cache_write = value
.get("prompt_tokens_details")
.and_then(|v| v.get("cache_write_tokens"))
.and_then(Value::as_i64)
.unwrap_or(0);
let input = (prompt_tokens - cache_read - cache_write).max(0);
let reasoning = value
.get("completion_tokens_details")
.and_then(|v| v.get("reasoning_tokens"))
.and_then(Value::as_i64);
Usage {
input,
output,
cache_read,
cache_write,
cache_write_1h: None,
reasoning,
total_tokens: input + output + cache_read + cache_write,
cost: Default::default(),
}
}
fn now_ms() -> i64 {
use std::time::{SystemTime, UNIX_EPOCH};
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|duration| duration.as_millis() as i64)
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::model::OpenaiCompletionsCompat;
use crate::types::{
ImageContent, ImageContentType, Message, Tool, ToolResultMessage, ToolResultRole,
UserMessage,
};
fn model() -> Model {
let mut model = Model::new(
"gpt-test",
"GPT Test",
Api::OpenaiCompletions,
"gateway",
"https://example.test/v1",
);
model.max_tokens = 1024;
model.reasoning = true;
model
}
#[test]
fn builds_chat_completions_request() {
let mut context = Context::new(vec![Message::User(UserMessage::new("hello", 0))]);
context.system_prompt = Some("system".into());
context.tools.push(Tool {
name: "read".into(),
description: "Read a file".into(),
parameters: json!({ "type": "object" }).into(),
constrained_sampling: None,
});
let mut opts = SimpleStreamOptions::default();
opts.reasoning = Some(ThinkingLevel::High);
let body = build_request(&model(), &context, &opts);
assert_eq!(body["model"], "gpt-test");
assert_eq!(body["messages"][0]["role"], "system");
assert_eq!(body["messages"][1]["role"], "user");
assert_eq!(body["tools"][0]["function"]["name"], "read");
assert_eq!(body["reasoning_effort"], "high");
assert_eq!(body["max_completion_tokens"], 1024);
assert_eq!(body["stream_options"]["include_usage"], true);
}
#[test]
fn applies_message_compat_and_preserves_tool_result_images() {
let mut model = model();
model.input.push(InputModality::Image);
model.compat = Some(StreamingProtocolCompat::OpenaiCompletions(
OpenaiCompletionsCompat {
supports_store: Some(true),
supports_developer_role: Some(true),
requires_tool_result_name: Some(true),
requires_assistant_after_tool_result: Some(true),
requires_thinking_as_text: Some(true),
..Default::default()
},
));
let mut assistant =
AssistantMessage::empty(Api::OpenaiCompletions, "gateway", "gpt-test", 1);
assistant.content = vec![Content::thinking("private"), Content::text("answer")];
let tool_result = ToolResultMessage {
role: ToolResultRole,
tool_call_id: "call-1".into(),
tool_name: "read".into(),
content: vec![Content::Image(ImageContent {
kind: ImageContentType,
data: "aW1hZ2U=".into(),
mime_type: "image/png".into(),
})],
details: None,
usage: None,
added_tool_names: vec![],
is_error: false,
timestamp: 2,
};
let context = Context {
system_prompt: Some("system".into()),
messages: vec![Message::Assistant(Box::new(assistant)), tool_result.into()],
tools: vec![],
};
let body = build_request(&model, &context, &SimpleStreamOptions::default());
assert_eq!(body["messages"][0]["role"], "developer");
assert_eq!(body["messages"][1]["content"][0]["text"], "private");
assert_eq!(body["messages"][1]["content"][1]["text"], "answer");
assert_eq!(body["messages"][2]["role"], "tool");
assert_eq!(body["messages"][2]["name"], "read");
assert_eq!(body["messages"][2]["content"], "(see attached image)");
assert_eq!(body["messages"][3]["role"], "assistant");
assert_eq!(body["messages"][4]["role"], "user");
assert_eq!(
body["messages"][4]["content"][1]["image_url"]["url"],
"data:image/png;base64,aW1hZ2U="
);
assert_eq!(body["store"], false);
}
#[test]
fn maps_text_tool_calls_and_usage() {
let model = model();
let (mut producer, _stream) = create_assistant_message_event_stream();
let mut state = StreamState::new(&model);
state.start(&mut producer);
state.apply_chunk(
&mut producer,
&json!({
"id": "chat-1",
"model": "gpt-test-actual",
"choices": [{"delta": {"content": "hello"}, "finish_reason": null}]
}),
);
state.apply_chunk(
&mut producer,
&json!({
"choices": [{"delta": {"tool_calls": [{
"index": 0,
"id": "call-1",
"function": {"name": "read", "arguments": "{\"path\":\"a\"}"}
}]}, "finish_reason": "tool_calls"}],
"usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15}
}),
);
state.finish(&mut producer, &model);
assert_eq!(state.output.stop_reason, StopReason::ToolUse);
assert_eq!(state.output.response_id.as_deref(), Some("chat-1"));
assert_eq!(state.output.usage.total_tokens, 15);
assert!(matches!(&state.output.content[0], Content::Text(text) if text.text == "hello"));
assert!(
matches!(&state.output.content[1], Content::ToolCall(call) if call.name == "read" && call.arguments["path"] == "a")
);
}
#[test]
fn splits_cached_usage_without_double_counting_prompt_tokens() {
let usage = parse_usage(&json!({
"prompt_tokens": 100,
"completion_tokens": 20,
"prompt_tokens_details": {
"cached_tokens": 40,
"cache_write_tokens": 10
},
"completion_tokens_details": {"reasoning_tokens": 5},
"total_tokens": 120
}));
assert_eq!(usage.input, 50);
assert_eq!(usage.cache_read, 40);
assert_eq!(usage.cache_write, 10);
assert_eq!(usage.output, 20);
assert_eq!(usage.reasoning, Some(5));
assert_eq!(usage.total_tokens, 120);
}
#[test]
fn maps_provider_error_finish_reason_to_error() {
let model = model();
let (mut producer, _stream) = create_assistant_message_event_stream();
let mut state = StreamState::new(&model);
state.start(&mut producer);
state.apply_chunk(
&mut producer,
&json!({
"choices": [{"delta": {}, "finish_reason": "content_filter"}]
}),
);
state.finish(&mut producer, &model);
assert_eq!(state.output.stop_reason, StopReason::Error);
assert_eq!(
state.output.error_message.as_deref(),
Some("Provider finish_reason: content_filter")
);
}
#[test]
fn missing_finish_reason_obeys_compat_flag() {
let mut model = model();
let (mut producer, _stream) = create_assistant_message_event_stream();
let mut state = StreamState::new(&model);
state.start(&mut producer);
state.finish(&mut producer, &model);
assert_eq!(state.output.stop_reason, StopReason::Error);
model.compat = Some(StreamingProtocolCompat::OpenaiCompletions(
OpenaiCompletionsCompat {
supports_finish_reason: Some(false),
..Default::default()
},
));
let (mut producer, _stream) = create_assistant_message_event_stream();
let mut state = StreamState::new(&model);
state.start(&mut producer);
state.finish(&mut producer, &model);
assert_eq!(state.output.stop_reason, StopReason::Stop);
}
#[test]
fn builds_endpoint_without_duplicate_v1() {
assert_eq!(
chat_completions_url("https://api.openai.com/v1/"),
"https://api.openai.com/v1/chat/completions"
);
assert_eq!(
chat_completions_url("https://gateway.test"),
"https://gateway.test/v1/chat/completions"
);
}
}