use crate::providers::{
error::ProviderError,
sse::{diagnostic_snippet, next_sse_event_boundary, sse_data},
};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::{BTreeMap, BTreeSet, btree_map::Entry};
mod chat_completions;
mod inline_tools;
mod openai_responses;
mod reasoning;
use chat_completions::chat_finish_reason;
use inline_tools::normalize_extra_quoted_tool_arguments;
use openai_responses::{
is_whole_response_completion, is_whole_response_failure, whole_response_failure_detail,
};
use reasoning::reasoning_summary_text;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub struct ToolCall {
pub id: String,
pub name: String,
pub arguments: Value,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct Usage {
pub input: u64,
pub output: u64,
pub cache_read: u64,
pub cache_write: u64,
pub total: u64,
pub reasoning_tokens: Option<u64>,
}
#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct UsagePresence {
pub(crate) input: bool,
pub(crate) output: bool,
pub(crate) cache_read: bool,
pub(crate) cache_write: bool,
pub(crate) total: bool,
pub(crate) reasoning: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub(crate) struct UsageObservation {
pub(crate) usage: Usage,
pub(crate) presence: UsagePresence,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ReasoningSummary {
pub text: String,
pub item_id: Option<String>,
pub turn_id: Option<String>,
#[serde(default)]
pub provider_summary: bool,
}
const MAX_SSE_EVENT_BUFFER_BYTES: usize = 1024 * 1024;
const MAX_TOOL_ARGUMENT_BYTES: usize = 1024 * 1024;
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
pub enum ProviderEvent {
TextDelta(String),
ReasoningSummaryDelta(String),
ReasoningSummaryComplete(String),
ReasoningSummaryCompleteIdentified(ReasoningSummary),
ToolCall(ToolCall),
ResponseItem(Value),
Usage(Usage),
UsagePartial(Usage),
UsageObserved(UsageObservation),
ServiceTier(String),
Done,
ResponseIdentity(crate::providers::error::ResponseAttemptIdentity),
}
pub(crate) fn is_semantic_progress_event(event: &ProviderEvent) -> bool {
match event {
ProviderEvent::TextDelta(_)
| ProviderEvent::ReasoningSummaryDelta(_)
| ProviderEvent::ReasoningSummaryComplete(_)
| ProviderEvent::ReasoningSummaryCompleteIdentified(_)
| ProviderEvent::ToolCall(_)
| ProviderEvent::Usage(_)
| ProviderEvent::UsagePartial(_)
| ProviderEvent::UsageObserved(_)
| ProviderEvent::Done => true,
ProviderEvent::ResponseItem(item) => {
matches!(
item.get("type").and_then(Value::as_str),
Some(
"function_call"
| "function_call_output"
| "reasoning"
| "thinking"
| "redacted_thinking"
)
) || item
.get("tool_calls")
.and_then(Value::as_array)
.is_some_and(|tool_calls| !tool_calls.is_empty())
}
ProviderEvent::ServiceTier(_) | ProviderEvent::ResponseIdentity(_) => false,
}
}
#[derive(Debug, Clone, Default, PartialEq)]
pub(crate) struct StreamParseOutcome {
pub(crate) events: Vec<ProviderEvent>,
pub(crate) semantic_progress: bool,
pub(crate) unsafe_recovery_progress: bool,
}
#[derive(Debug, Default)]
pub(crate) struct StreamParser {
event_buffer: String,
tool_calls: BTreeMap<String, PendingToolCall>,
chat_tool_call_indices: BTreeMap<String, String>,
emitted_response_item_keys: BTreeSet<String>,
reasoning_summary_text: String,
saw_raw_reasoning: bool,
completed_reasoning_summary_text: Option<String>,
completed_reasoning_summary_keys: BTreeMap<String, String>,
saw_identified_reasoning_completion: bool,
emitted_text_delta: bool,
saw_terminal_completion: bool,
emitted_done: bool,
unsafe_chat_tool_call_completion: bool,
next_tool_call_sequence: u64,
content_buffer: String,
content_search_start: usize,
thinking_complete: bool,
gemma_inline_tool_calls_enabled: bool,
gemma_inline_tool_call_counter: u64,
response_model: Option<String>,
returned_service_tier: Option<String>,
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
struct PendingToolCall {
call_id: Option<String>,
name: Option<String>,
arguments_text: String,
emitted: bool,
provider_index: Option<u64>,
first_seen_sequence: u64,
source: ToolCallSource,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
enum ToolCallSource {
#[default]
Responses,
ChatCompletions,
}
fn gemma_inline_tool_calls_enabled(provider_id: &str, model: &str) -> bool {
let provider_id = provider_id.to_ascii_lowercase();
let model = model.to_ascii_lowercase();
let custom_vllm_profile = provider_id.contains("vllm") || provider_id.contains("foundry");
custom_vllm_profile && (model.contains("gemma") || model.contains("diffusiongemma"))
}
impl StreamParser {
pub(crate) fn for_provider_model(provider_id: &str, model: &str) -> Self {
Self {
gemma_inline_tool_calls_enabled: gemma_inline_tool_calls_enabled(provider_id, model),
..Self::default()
}
}
#[cfg(test)]
pub(crate) fn push_chunk(&mut self, chunk: &str) -> anyhow::Result<Vec<ProviderEvent>> {
Ok(self.push_chunk_outcome(chunk)?.events)
}
pub(crate) fn push_chunk_outcome(&mut self, chunk: &str) -> anyhow::Result<StreamParseOutcome> {
self.event_buffer.push_str(chunk);
let mut buffer = std::mem::take(&mut self.event_buffer);
let mut events = Vec::new();
let mut semantic_progress = false;
let mut unsafe_recovery_progress = false;
while let Some((boundary, boundary_len)) = next_sse_event_boundary(&buffer) {
let parsed = sse_data(&buffer[..boundary]).map(|data| self.parse_data_event(&data));
buffer.drain(..boundary + boundary_len);
if let Some(parsed) = parsed {
let parsed = match parsed {
Ok(parsed) => parsed,
Err(error) => {
self.event_buffer = buffer;
return Err(error);
}
};
semantic_progress |= parsed.semantic_progress
|| parsed.events.iter().any(is_semantic_progress_event);
unsafe_recovery_progress |= parsed.unsafe_recovery_progress;
events.extend(parsed.events);
}
}
self.event_buffer = buffer;
self.ensure_event_buffer_within_limit()?;
Ok(StreamParseOutcome {
events,
semantic_progress,
unsafe_recovery_progress,
})
}
pub(crate) fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
if !self.event_buffer.trim().is_empty() {
let raw_event = std::mem::take(&mut self.event_buffer);
if let Some(data) = sse_data(&raw_event)
&& data == "[DONE]"
{
self.saw_terminal_completion = true;
return self.done_event(true);
}
anyhow::bail!(
"provider SSE stream ended with incomplete event buffer: {}",
diagnostic_snippet(&raw_event)
);
}
for (key, pending) in &self.tool_calls {
if !pending.emitted && !pending.arguments_text.trim().is_empty() {
parse_arguments_text(&pending.arguments_text).map_err(|error| {
anyhow::anyhow!(
"provider SSE stream ended with incomplete tool call arguments for {key}: {error}: {}",
diagnostic_snippet(&pending.arguments_text)
)
})?;
}
}
if !self.saw_terminal_completion {
return Err(ProviderError::stream_terminal(
"missing provider stream completion before EOF",
)
.into());
}
Ok(Vec::new())
}
fn parse_data_event(&mut self, data: &str) -> anyhow::Result<StreamParseOutcome> {
let mut events = Vec::new();
let mut semantic_progress = false;
if data == "[DONE]" {
semantic_progress |= !self.saw_terminal_completion;
self.saw_terminal_completion = true;
events.extend(self.done_event(true)?);
return Ok(StreamParseOutcome {
events,
semantic_progress,
unsafe_recovery_progress: false,
});
}
let value = serde_json::from_str::<Value>(data).map_err(|error| {
anyhow::anyhow!(
"malformed provider SSE data JSON: {error}: {}",
diagnostic_snippet(data)
)
})?;
let item_type = value
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
self.response_model = self.response_model.clone().or_else(|| {
value
.pointer("/response/model")
.and_then(Value::as_str)
.map(crate::providers::error::bounded_response_identity_string)
.or_else(|| {
value
.get("model")
.and_then(Value::as_str)
.map(crate::providers::error::bounded_response_identity_string)
})
});
let returned_tier = value
.get("service_tier")
.or_else(|| value.pointer("/response/service_tier"))
.and_then(crate::fast::parse_returned_service_tier);
if let Some(tier) = returned_tier
&& self.returned_service_tier.as_deref() != Some(tier.as_str())
{
self.returned_service_tier = Some(tier.clone());
events.push(ProviderEvent::ServiceTier(tier));
}
if is_whole_response_failure(&value, item_type) {
let message = match whole_response_failure_detail(&value) {
Some(detail) => format!(
"provider stream ended with failed or incomplete response: {detail} | raw: {}",
diagnostic_snippet(data)
),
None => format!(
"provider stream ended with failed or incomplete response: {}",
diagnostic_snippet(data)
),
};
return Err(ProviderError::stream_failed_incomplete(message).into());
}
let chat_finish_reason = chat_finish_reason(&value);
if matches!(
chat_finish_reason,
Some(reason) if !matches!(reason, "stop" | "tool_calls")
) {
self.unsafe_chat_tool_call_completion = true;
}
let chat_finish_is_terminal = matches!(
chat_finish_reason,
Some("stop" | "tool_calls" | "length" | "content_filter")
);
let is_terminal_completion =
is_whole_response_completion(&value, item_type) || chat_finish_is_terminal;
if let Some((summary, provider_summary)) =
self.reasoning_summary_delta_from_event(&value, item_type)
{
self.reasoning_summary_text.push_str(summary);
self.saw_raw_reasoning |= !provider_summary;
semantic_progress = true;
events.push(ProviderEvent::ReasoningSummaryDelta(summary.to_string()));
}
if let Some((summary, item_id)) = self.reasoning_summary_done_from_event(&value, item_type)
{
events.extend(self.reconcile_reasoning_summary_complete(summary, item_id));
}
if !matches!(
item_type,
"response.function_call_arguments.delta"
| "response.reasoning_summary_text.delta"
| "response.reasoning_summary_text.done"
) {
if let Some(delta) = self.chat_content_delta_from_event(&value) {
let (reasoning_delta, text_delta) = self.process_chat_content_delta(delta);
if let Some(reasoning_delta) = reasoning_delta {
self.reasoning_summary_text.push_str(&reasoning_delta);
self.saw_raw_reasoning = true;
semantic_progress = true;
events.push(ProviderEvent::ReasoningSummaryDelta(reasoning_delta));
}
if let Some(text_delta) = text_delta {
self.emitted_text_delta = true;
semantic_progress = true;
events.push(ProviderEvent::TextDelta(text_delta));
}
} else if let Some(delta) = self.text_delta_from_event(&value, item_type) {
self.emitted_text_delta = true;
semantic_progress = true;
events.push(ProviderEvent::TextDelta(delta.to_string()));
}
}
let response_items = self.parse_response_items(&value, item_type);
for item in response_items {
if let Some(summary) = reasoning_summary_text(&item) {
events.extend(self.reconcile_reasoning_summary_complete(
&summary,
item.get("id").and_then(Value::as_str),
));
}
events.push(ProviderEvent::ResponseItem(item));
}
let (tool_calls, tool_call_progress) = self.parse_tool_calls(&value)?;
semantic_progress |= tool_call_progress;
if matches!(chat_finish_reason, Some("tool_calls" | "stop")) {
self.flush_pending_chat_content(&mut events);
self.emit_completed_chat_tool_calls(&mut events)?;
}
events.extend(tool_calls.into_iter().map(ProviderEvent::ToolCall));
if let Some(parsed_usage) = parse_usage(&value) {
events.push(ProviderEvent::UsageObserved(UsageObservation {
usage: parsed_usage.usage,
presence: parsed_usage.presence,
}));
}
if is_terminal_completion {
self.flush_pending_chat_content(&mut events);
self.saw_terminal_completion = true;
semantic_progress = true;
events.extend(self.done_event(false)?);
}
semantic_progress |= events.iter().any(is_semantic_progress_event);
Ok(StreamParseOutcome {
events,
semantic_progress,
unsafe_recovery_progress: tool_call_progress,
})
}
fn done_event(&mut self, complete_chat_tool_calls: bool) -> anyhow::Result<Vec<ProviderEvent>> {
let mut events = Vec::new();
self.flush_pending_chat_content(&mut events);
if complete_chat_tool_calls && !self.unsafe_chat_tool_call_completion {
self.emit_completed_chat_tool_calls(&mut events)?;
}
if !self.saw_identified_reasoning_completion
&& !self.reasoning_summary_text.trim().is_empty()
&& self.completed_reasoning_summary_text.as_deref()
!= Some(self.reasoning_summary_text.as_str())
{
self.completed_reasoning_summary_text = Some(self.reasoning_summary_text.clone());
if self.saw_raw_reasoning {
events.push(ProviderEvent::ReasoningSummaryComplete(
self.reasoning_summary_text.clone(),
));
} else {
events.push(ProviderEvent::ReasoningSummaryCompleteIdentified(
ReasoningSummary {
text: self.reasoning_summary_text.clone(),
item_id: None,
turn_id: None,
provider_summary: true,
},
));
}
}
if !self.emitted_done {
self.emitted_done = true;
events.push(ProviderEvent::Done);
}
Ok(events)
}
fn parse_tool_calls(&mut self, value: &Value) -> anyhow::Result<(Vec<ToolCall>, bool)> {
let mut calls = Vec::new();
let mut semantic_progress = false;
semantic_progress |= self.parse_response_tool_calls(value, &mut calls)?;
semantic_progress |= self.parse_chat_tool_calls(value, &mut calls)?;
Ok((calls, semantic_progress))
}
fn pending_for_key(
&mut self,
key: &str,
provider_index: Option<u64>,
source: ToolCallSource,
) -> &mut PendingToolCall {
match self.tool_calls.entry(key.to_string()) {
Entry::Occupied(entry) => {
let pending = entry.into_mut();
if pending.provider_index.is_none() {
pending.provider_index = provider_index;
}
if pending.source != source {
pending.source = source;
}
pending
}
Entry::Vacant(entry) => {
let sequence = self.next_tool_call_sequence;
self.next_tool_call_sequence += 1;
entry.insert(PendingToolCall {
provider_index,
first_seen_sequence: sequence,
source,
..PendingToolCall::default()
})
}
}
}
fn migrate_pending_tool_call(&mut self, from_key: &str, to_key: &str) -> anyhow::Result<bool> {
if from_key == to_key {
return Ok(false);
}
let Some(from_pending) = self.tool_calls.remove(from_key) else {
return Ok(false);
};
match self.tool_calls.entry(to_key.to_string()) {
Entry::Vacant(entry) => {
entry.insert(from_pending);
}
Entry::Occupied(mut entry) => {
merge_pending_tool_call(entry.get_mut(), from_pending, to_key)?;
}
}
Ok(true)
}
fn ensure_event_buffer_within_limit(&self) -> anyhow::Result<()> {
if self.event_buffer.len() <= MAX_SSE_EVENT_BUFFER_BYTES {
return Ok(());
}
Err(ProviderError::stream_terminal(format!(
"provider SSE event exceeded maximum buffered size of {MAX_SSE_EVENT_BUFFER_BYTES} bytes before a frame boundary"
))
.into())
}
fn push_tool_arguments_delta(
pending: &mut PendingToolCall,
delta: &str,
) -> anyhow::Result<bool> {
if delta.is_empty() {
return Ok(false);
}
let next_len = pending.arguments_text.len().saturating_add(delta.len());
if next_len > MAX_TOOL_ARGUMENT_BYTES {
return Err(ProviderError::stream_terminal(format!(
"provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
))
.into());
}
pending.arguments_text.push_str(delta);
Ok(true)
}
fn set_tool_arguments_text(
pending: &mut PendingToolCall,
arguments_text: String,
) -> anyhow::Result<bool> {
if arguments_text.len() > MAX_TOOL_ARGUMENT_BYTES {
return Err(ProviderError::stream_terminal(format!(
"provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
))
.into());
}
if pending.arguments_text == arguments_text {
return Ok(false);
}
pending.arguments_text = arguments_text;
Ok(true)
}
pub(crate) fn response_model(&self) -> Option<String> {
self.response_model.clone()
}
}
fn merge_pending_tool_call(
target: &mut PendingToolCall,
source: PendingToolCall,
key: &str,
) -> anyhow::Result<()> {
if let Some(call_id) = source.call_id {
if let Some(existing) = &target.call_id
&& existing != &call_id
{
anyhow::bail!(
"conflicting duplicate provider tool call id for {key}: {existing} vs {call_id}"
);
}
target.call_id = Some(call_id);
}
if let Some(name) = source.name {
if let Some(existing) = &target.name
&& existing != &name
{
anyhow::bail!(
"conflicting duplicate provider tool call name for {key}: {existing} vs {name}"
);
}
target.name = Some(name);
}
if !source.arguments_text.is_empty() {
let target_arguments = std::mem::take(&mut target.arguments_text);
target.arguments_text = source.arguments_text;
let next_len = target
.arguments_text
.len()
.saturating_add(target_arguments.len());
if next_len > MAX_TOOL_ARGUMENT_BYTES {
return Err(ProviderError::stream_terminal(format!(
"provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
))
.into());
}
target.arguments_text.push_str(&target_arguments);
}
target.emitted |= source.emitted;
target.provider_index = target.provider_index.or(source.provider_index);
target.first_seen_sequence = target.first_seen_sequence.min(source.first_seen_sequence);
target.source = source_priority(target.source, source.source);
Ok(())
}
fn source_priority(left: ToolCallSource, right: ToolCallSource) -> ToolCallSource {
if matches!(left, ToolCallSource::ChatCompletions)
|| matches!(right, ToolCallSource::ChatCompletions)
{
ToolCallSource::ChatCompletions
} else {
ToolCallSource::Responses
}
}
fn arguments_as_text(arguments: &Value) -> String {
match arguments {
Value::String(text) => text.clone(),
value => value.to_string(),
}
}
fn parse_arguments_text(text: &str) -> anyhow::Result<Value> {
if text.trim().is_empty() {
return Ok(Value::Object(Default::default()));
}
parse_arguments_json_value(text)
.map(normalize_extra_quoted_tool_arguments)
.map_err(|error| {
anyhow::anyhow!(
"malformed non-empty provider tool call arguments: {error}: {}",
diagnostic_snippet(text)
)
})
}
fn parse_arguments_json_value(text: &str) -> serde_json::Result<Value> {
match serde_json::from_str::<Value>(text) {
Ok(value) => Ok(value),
Err(strict_error) => {
let mut stream = serde_json::Deserializer::from_str(text).into_iter::<Value>();
let value = match stream.next() {
Some(Ok(value)) => value,
Some(Err(error)) => return Err(error),
None => return Err(strict_error),
};
let trailing = text[stream.byte_offset()..].trim();
if trailing.is_empty() || trailing_is_empty_json_objects(trailing) {
Ok(value)
} else {
Err(strict_error)
}
}
}
}
fn trailing_is_empty_json_objects(mut text: &str) -> bool {
loop {
text = text.trim_start();
if text.is_empty() {
return true;
}
let Some(rest) = text.strip_prefix("{}") else {
return false;
};
text = rest;
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub(crate) struct ParsedUsage {
pub(crate) usage: Usage,
pub(crate) input_tokens: Option<u64>,
pub(crate) reasoning_tokens: Option<u64>,
pub(crate) presence: UsagePresence,
}
fn parse_usage(value: &Value) -> Option<ParsedUsage> {
let usage = value
.pointer("/usage")
.filter(|usage| usage.is_object())
.or_else(|| value.pointer("/response/usage"))?
.as_object()?;
let input_tokens = usage
.get("input_tokens")
.and_then(Value::as_u64)
.or_else(|| usage.get("prompt_tokens").and_then(Value::as_u64));
let output_tokens = usage
.get("output_tokens")
.and_then(Value::as_u64)
.or_else(|| usage.get("completion_tokens").and_then(Value::as_u64));
let cache_read_tokens = usage
.get("input_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(Value::as_u64)
.or_else(|| {
usage
.get("prompt_tokens_details")
.and_then(|d| d.get("cached_tokens"))
.and_then(Value::as_u64)
});
let cache_write_tokens = usage
.get("cache_write_tokens")
.and_then(Value::as_u64)
.or_else(|| {
usage
.get("input_tokens_details")
.and_then(|d| d.get("cache_write_tokens"))
.and_then(Value::as_u64)
})
.or_else(|| {
usage
.get("prompt_tokens_details")
.and_then(|d| d.get("cache_write_tokens"))
.and_then(Value::as_u64)
});
let total_tokens = usage.get("total_tokens").and_then(Value::as_u64);
let reasoning_tokens = usage
.get("output_tokens_details")
.and_then(|d| d.get("reasoning_tokens"))
.and_then(Value::as_u64)
.or_else(|| {
usage
.get("completion_tokens_details")
.and_then(|d| d.get("reasoning_tokens"))
.and_then(Value::as_u64)
});
if [
input_tokens,
output_tokens,
cache_read_tokens,
cache_write_tokens,
total_tokens,
reasoning_tokens,
]
.iter()
.all(Option::is_none)
{
return None;
}
let input = input_tokens.unwrap_or_default();
let output = output_tokens.unwrap_or_default();
let cache_read = cache_read_tokens.unwrap_or_default();
let cache_write = cache_write_tokens.unwrap_or_default();
let total = total_tokens.unwrap_or_else(|| input.saturating_add(output));
Some(ParsedUsage {
usage: Usage {
input,
output,
cache_read,
cache_write,
total,
reasoning_tokens,
},
input_tokens,
reasoning_tokens,
presence: UsagePresence {
input: input_tokens.is_some(),
output: output_tokens.is_some(),
cache_read: cache_read_tokens.is_some(),
cache_write: cache_write_tokens.is_some(),
total: total_tokens.is_some(),
reasoning: reasoning_tokens.is_some(),
},
})
}