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"))
}
const MAX_WHOLE_RESPONSE_FAILURE_DETAIL_CHARS: usize = 400;
pub(super) fn whole_response_failure_detail(value: &Value) -> Option<String> {
let response_error = value.pointer("/response/error");
let code = response_error
.and_then(|error| error.get("code"))
.and_then(Value::as_str)
.or_else(|| value.pointer("/error/code").and_then(Value::as_str))
.map(crate::output::redact_sensitive_text);
let message = response_error
.and_then(|error| error.get("message"))
.and_then(Value::as_str)
.or_else(|| value.pointer("/error/message").and_then(Value::as_str))
.map(crate::output::redact_sensitive_text);
let param = response_error
.and_then(|error| error.get("param"))
.and_then(Value::as_str)
.map(crate::output::redact_sensitive_text);
let incomplete_reason = value
.pointer("/response/incomplete_details/reason")
.and_then(Value::as_str)
.map(crate::output::redact_sensitive_text);
let mut parts = Vec::new();
if let Some(code) = code {
parts.push(format!("code={code}"));
}
if let Some(message) = message {
parts.push(format!("message={message:?}"));
}
if let Some(param) = param {
parts.push(format!("param={param}"));
}
if let Some(reason) = incomplete_reason {
parts.push(format!("incomplete_reason={reason}"));
}
let detail = parts.join(" ");
(!detail.is_empty()).then(|| {
detail
.chars()
.take(MAX_WHOLE_RESPONSE_FAILURE_DETAIL_CHARS)
.collect()
})
}
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}"))
})
}
#[cfg(test)]
mod tests {
use super::*;
use serde_json::json;
#[test]
fn whole_response_failure_detail_extracts_error_fields() {
let detail = whole_response_failure_detail(&json!({
"response": {
"error": {
"code": "server_error",
"message": "upstream unavailable",
"param": "input"
}
}
}))
.expect("response failure detail");
assert_eq!(
detail,
r#"code=server_error message="upstream unavailable" param=input"#
);
}
#[test]
fn whole_response_failure_detail_extracts_incomplete_reason() {
assert_eq!(
whole_response_failure_detail(&json!({
"response": {
"incomplete_details": {"reason": "max_output_tokens"}
}
})),
Some("incomplete_reason=max_output_tokens".to_string())
);
}
#[test]
fn whole_response_failure_detail_falls_back_to_top_level_error() {
assert_eq!(
whole_response_failure_detail(&json!({
"error": {"code": "rate_limit", "message": "try again later"}
})),
Some(r#"code=rate_limit message="try again later""#.to_string())
);
}
#[test]
fn whole_response_failure_detail_returns_none_without_failure_fields() {
assert_eq!(
whole_response_failure_detail(&json!({
"type": "response.completed",
"response": {"status": "completed"}
})),
None
);
}
#[test]
fn whole_response_failure_detail_redacts_sensitive_text_and_is_bounded() {
let detail = whole_response_failure_detail(&json!({
"response": {
"error": {
"code": "server_error",
"message": format!("Bearer abcdefghijk {}", "é".repeat(500))
}
}
}))
.expect("response failure detail");
assert!(detail.contains("Bearer <redacted>"), "{detail}");
assert!(!detail.contains("abcdefghijk"), "{detail}");
assert!(detail.chars().count() <= MAX_WHOLE_RESPONSE_FAILURE_DETAIL_CHARS);
}
}