use super::*;
use serde_json::Value;
impl StreamParser {
pub(super) fn chat_content_delta_from_event<'a>(&self, value: &'a Value) -> Option<&'a str> {
value
.pointer("/choices/0/delta/content")
.or_else(|| value.pointer("/choices/0/message/content"))
.and_then(Value::as_str)
}
pub(super) fn complete_pending_chat_tool_calls(
&mut self,
) -> anyhow::Result<(Vec<ToolCall>, bool)> {
let mut pending_keys = self
.tool_calls
.iter()
.filter(|(_, pending)| {
!pending.emitted && pending.source == ToolCallSource::ChatCompletions
})
.map(|(key, pending)| {
(
key.clone(),
pending.provider_index.unwrap_or(u64::MAX),
pending.first_seen_sequence,
)
})
.collect::<Vec<_>>();
pending_keys.sort_by_key(|(_, provider_index, sequence)| (*provider_index, *sequence));
let mut calls = Vec::new();
let mut semantic_progress = false;
for (key, _, _) in pending_keys {
let Some(pending) = self.tool_calls.get_mut(&key) else {
continue;
};
if let (Some(call_id), Some(name)) = (pending.call_id.clone(), pending.name.clone()) {
let arguments = parse_arguments_text(&pending.arguments_text)?;
pending.emitted = true;
semantic_progress = true;
calls.push(ToolCall {
id: call_id,
name,
arguments,
});
}
}
Ok((calls, semantic_progress))
}
pub(super) fn emit_completed_chat_tool_calls(
&mut self,
events: &mut Vec<ProviderEvent>,
) -> anyhow::Result<bool> {
let (tool_calls, progress) = self.complete_pending_chat_tool_calls()?;
if !tool_calls.is_empty() {
events.push(ProviderEvent::ResponseItem(chat_tool_call_response_item(
&tool_calls,
)));
events.extend(tool_calls.into_iter().map(ProviderEvent::ToolCall));
}
Ok(progress)
}
pub(super) fn collect_legacy_function_call(
&mut self,
function_call: &Value,
calls: &mut Vec<ToolCall>,
event_is_complete: bool,
raw_arguments_can_complete: bool,
) -> anyhow::Result<bool> {
let key = "chat_legacy_function_call";
let call_id = function_call
.get("id")
.or_else(|| function_call.get("call_id"))
.and_then(Value::as_str)
.unwrap_or("call_legacy_function_call");
let name = function_call.get("name").and_then(Value::as_str);
let raw_arguments = function_call.get("arguments");
let pending_created = !self.tool_calls.contains_key(key);
let pending = self.pending_for_key(key, None, ToolCallSource::ChatCompletions);
let mut semantic_progress = pending_created;
if pending.call_id.as_deref() != Some(call_id) {
pending.call_id = Some(call_id.to_string());
semantic_progress = true;
}
if let Some(name) = name {
if let Some(existing) = &pending.name
&& existing != name
{
anyhow::bail!(
"conflicting duplicate provider tool call name for {key}: {existing} vs {name}"
);
}
if pending.name.as_deref() != Some(name) {
pending.name = Some(name.to_string());
semantic_progress = true;
}
}
if let Some(arguments) = raw_arguments {
let arguments_text = arguments_as_text(arguments);
if raw_arguments_can_complete {
if Self::set_tool_arguments_text(pending, arguments_text)? {
semantic_progress = true;
}
} else if Self::push_tool_arguments_delta(pending, &arguments_text)? {
semantic_progress = true;
}
}
let is_complete =
event_is_complete || (raw_arguments_can_complete && raw_arguments.is_some());
if is_complete
&& !pending.emitted
&& let (Some(call_id), Some(name)) = (pending.call_id.clone(), pending.name.clone())
{
let arguments = parse_arguments_text(&pending.arguments_text)?;
pending.emitted = true;
semantic_progress = true;
calls.push(ToolCall {
id: call_id,
name,
arguments,
});
}
Ok(semantic_progress)
}
pub(super) fn parse_chat_tool_calls(
&mut self,
value: &Value,
calls: &mut Vec<ToolCall>,
) -> anyhow::Result<bool> {
let mut semantic_progress = false;
if let Some(tool_calls) = value
.pointer("/choices/0/delta/tool_calls")
.and_then(Value::as_array)
{
for item in tool_calls {
semantic_progress |= self.collect_chat_tool_call(item, calls, false, false)?;
}
}
if let Some(tool_calls) = value
.pointer("/choices/0/message/tool_calls")
.and_then(Value::as_array)
{
for item in tool_calls {
semantic_progress |= self.collect_chat_tool_call(item, calls, true, true)?;
}
}
if let Some(function_call) = value.pointer("/choices/0/delta/function_call") {
semantic_progress |=
self.collect_legacy_function_call(function_call, calls, false, false)?;
}
if let Some(function_call) = value.pointer("/choices/0/message/function_call") {
semantic_progress |=
self.collect_legacy_function_call(function_call, calls, true, true)?;
}
Ok(semantic_progress)
}
pub(super) fn collect_chat_tool_call(
&mut self,
value: &Value,
calls: &mut Vec<ToolCall>,
event_is_complete: bool,
raw_arguments_can_complete: bool,
) -> anyhow::Result<bool> {
let mut semantic_progress = false;
let item_type = value
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
let function = value.get("function");
let item_id = value
.get("item_id")
.or_else(|| value.get("id"))
.and_then(Value::as_str);
let call_id = value
.get("call_id")
.or_else(|| function.and_then(|function| function.get("call_id")))
.and_then(Value::as_str);
let provider_index = value.get("index").and_then(Value::as_u64);
let index_key = provider_index.map(|index| format!("chat_index:{index}"));
let key = item_id
.map(ToString::to_string)
.or_else(|| call_id.map(ToString::to_string))
.or_else(|| {
index_key
.as_ref()
.and_then(|index| self.chat_tool_call_indices.get(index).cloned())
})
.or_else(|| index_key.clone());
if let (Some(index_key), Some(key)) = (&index_key, &key)
&& (item_id.is_some() || call_id.is_some())
{
let index_pending_key = index_key.clone();
if index_pending_key != *key && self.tool_calls.contains_key(&index_pending_key) {
semantic_progress |= self.migrate_pending_tool_call(&index_pending_key, key)?;
}
match self.chat_tool_call_indices.entry(index_key.clone()) {
Entry::Vacant(entry) => {
entry.insert(key.clone());
semantic_progress = true;
}
Entry::Occupied(_) => {}
}
}
let name = value
.get("name")
.or_else(|| function.and_then(|function| function.get("name")))
.and_then(Value::as_str);
let raw_arguments = value
.get("arguments")
.or_else(|| function.and_then(|function| function.get("arguments")));
if matches!(item_type, "response.function_call_arguments.delta") {
if let (Some(key), Some(delta)) = (key, value.get("delta").and_then(Value::as_str)) {
let pending_created = !self.tool_calls.contains_key(&key);
let pending =
self.pending_for_key(&key, provider_index, ToolCallSource::ChatCompletions);
semantic_progress |= pending_created;
if Self::push_tool_arguments_delta(pending, delta)? {
semantic_progress = true;
}
}
return Ok(semantic_progress);
}
if let Some(key) = key {
let pending_created = !self.tool_calls.contains_key(&key);
let pending =
self.pending_for_key(&key, provider_index, ToolCallSource::ChatCompletions);
semantic_progress |= pending_created;
if let Some(call_id) = call_id {
if let Some(existing) = &pending.call_id
&& existing != call_id
{
anyhow::bail!(
"conflicting duplicate provider tool call id for {key}: {existing} vs {call_id}"
);
}
if pending.call_id.as_deref() != Some(call_id) {
pending.call_id = Some(call_id.to_string());
semantic_progress = true;
}
}
if pending.call_id.is_none() && (item_id.is_some() || key.starts_with("call_")) {
pending.call_id = Some(key.to_string());
semantic_progress = true;
}
if let Some(name) = name {
if let Some(existing) = &pending.name
&& existing != name
{
anyhow::bail!(
"conflicting duplicate provider tool call name for {key}: {existing} vs {name}"
);
}
if pending.name.as_deref() != Some(name) {
pending.name = Some(name.to_string());
semantic_progress = true;
}
}
if let Some(arguments) = raw_arguments {
let arguments_text = arguments_as_text(arguments);
if raw_arguments_can_complete {
if Self::set_tool_arguments_text(pending, arguments_text)? {
semantic_progress = true;
}
} else if Self::push_tool_arguments_delta(pending, &arguments_text)? {
semantic_progress = true;
}
}
let is_complete =
event_is_complete || (raw_arguments_can_complete && raw_arguments.is_some());
if is_complete
&& !pending.emitted
&& let (Some(call_id), Some(name)) = (pending.call_id.clone(), pending.name.clone())
{
let arguments = parse_arguments_text(&pending.arguments_text)?;
pending.emitted = true;
semantic_progress = true;
calls.push(ToolCall {
id: call_id,
name,
arguments,
});
}
}
Ok(semantic_progress)
}
}
pub(super) fn chat_finish_reason(value: &Value) -> Option<&str> {
value
.pointer("/choices/0/finish_reason")
.and_then(Value::as_str)
}
pub(super) fn chat_tool_call_response_item(tool_calls: &[ToolCall]) -> Value {
let tool_calls = tool_calls
.iter()
.map(|call| {
serde_json::json!({
"id": call.id,
"type": "function",
"function": {
"name": call.name,
"arguments": call.arguments.to_string(),
}
})
})
.collect::<Vec<_>>();
serde_json::json!({
"role": "assistant",
"content": null,
"tool_calls": tool_calls,
})
}