use super::*;
use serde_json::Value;
impl StreamParser {
pub(super) fn text_delta_from_event<'a>(
&self,
value: &'a Value,
item_type: &str,
) -> Option<&'a str> {
value
.pointer("/delta")
.or_else(|| value.pointer("/response/output_text/delta"))
.or_else(|| value.pointer("/delta/text"))
.and_then(Value::as_str)
.or_else(|| {
if self.emitted_text_delta || is_final_text_event(item_type) {
None
} else {
value.pointer("/text").and_then(Value::as_str)
}
})
}
pub(super) fn parse_response_items(&mut self, value: &Value, item_type: &str) -> Vec<Value> {
let mut items = Vec::new();
if matches!(
item_type,
"response.output_item.done" | "response.function_call_arguments.done"
) && let Some(item) = value.pointer("/item")
{
self.collect_response_item(item, &mut items);
}
if let Some(output) = value.pointer("/response/output").and_then(Value::as_array) {
for item in output {
self.collect_response_item(item, &mut items);
}
}
if let Some(output) = value.pointer("/output").and_then(Value::as_array) {
for item in output {
self.collect_response_item(item, &mut items);
}
}
self.collect_response_item(value, &mut items);
items
}
pub(super) fn collect_response_item(&mut self, value: &Value, items: &mut Vec<Value>) {
let item_type = value
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
if !matches!(item_type, "function_call" | "reasoning") {
return;
}
if item_type == "reasoning" {
let Some(key) = response_item_key(value) else {
return;
};
let summary = reasoning_summary_text(value).unwrap_or_default();
let changed = self
.completed_reasoning_summary_keys
.get(&key)
.is_none_or(|previous| previous != &summary);
if changed {
items.push(value.clone());
}
return;
}
let Some(key) = response_item_key(value) else {
return;
};
if self.emitted_response_item_keys.insert(key) {
items.push(value.clone());
}
}
pub(super) fn parse_response_tool_calls(
&mut self,
value: &Value,
calls: &mut Vec<ToolCall>,
) -> anyhow::Result<bool> {
let mut semantic_progress = false;
let event_type = value
.get("type")
.and_then(Value::as_str)
.unwrap_or_default();
let event_is_complete = matches!(
event_type,
"response.output_item.done" | "response.function_call_arguments.done"
);
semantic_progress |=
self.collect_response_tool_call(value, calls, event_is_complete, true)?;
if let Some(item) = value.pointer("/item") {
semantic_progress |= self.collect_response_tool_call(
item,
calls,
event_is_complete,
!matches!(event_type, "response.output_item.added"),
)?;
}
if let Some(output) = value.pointer("/response/output").and_then(Value::as_array) {
for item in output {
semantic_progress |= self.collect_response_tool_call(item, calls, true, true)?;
}
}
if let Some(output) = value.pointer("/output").and_then(Value::as_array) {
for item in output {
semantic_progress |= self.collect_response_tool_call(item, calls, true, true)?;
}
}
if let Some(tool_calls) = value.pointer("/tool_calls").and_then(Value::as_array) {
for item in tool_calls {
semantic_progress |= self.collect_response_tool_call(item, calls, true, true)?;
}
}
Ok(semantic_progress)
}
pub(super) fn collect_response_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::Responses);
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::Responses);
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() && 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 is_final_text_event(item_type: &str) -> bool {
matches!(
item_type,
"response.output_text.done" | "response.output_item.done" | "response.completed"
)
}
pub(super) fn is_whole_response_completion(value: &Value, item_type: &str) -> bool {
item_type == "response.completed" || response_status(value) == Some("completed")
}
pub(super) fn is_whole_response_failure(value: &Value, item_type: &str) -> bool {
matches!(item_type, "response.failed" | "response.incomplete")
|| matches!(response_status(value), Some("failed" | "incomplete"))
}
pub(super) fn response_status(value: &Value) -> Option<&str> {
value
.pointer("/response/status")
.and_then(Value::as_str)
.or_else(|| {
let is_response_object = matches!(
value.get("type").and_then(Value::as_str),
Some("response" | "response.completed" | "response.failed" | "response.incomplete")
) || value.get("object").and_then(Value::as_str)
== Some("response");
is_response_object
.then(|| value.get("status").and_then(Value::as_str))
.flatten()
})
}
pub(super) fn response_item_key(item: &Value) -> Option<String> {
item.get("id")
.and_then(Value::as_str)
.map(|id| format!("id:{id}"))
.or_else(|| {
item.get("call_id")
.and_then(Value::as_str)
.map(|call_id| format!("call:{call_id}"))
})
}