use serde_json::{Value, json};
use crate::ir::IrStreamEvent;
use crate::map::MapError;
use super::usage;
use super::{MAX_TOOL_CALL_INDEX, RawSse, check_index, protocol, str_field};
pub(super) fn failure_object(value: &Value) -> Option<&Value> {
fn useful(err: &Value) -> bool {
if !err.is_object() {
return false;
}
let message = err
.get("message")
.and_then(Value::as_str)
.is_some_and(|text| !text.trim().is_empty());
let code = match err.get("code") {
Some(Value::String(text)) => !text.trim().is_empty(),
Some(Value::Number(_)) => true,
_ => false,
};
message || code
}
value
.pointer("/response/error")
.filter(|err| useful(err))
.or_else(|| {
value
.pointer("/response/last_error")
.filter(|err| useful(err))
})
}
fn last_error_code(last: &Value) -> Option<String> {
let code = last.get("code")?;
if let Some(text) = code.as_str() {
let text = text.trim();
if text.is_empty() {
return None;
}
return Some(text.to_string());
}
code.as_i64().map(|n| n.to_string())
}
pub(super) fn decode(name: &str, value: &Value) -> Result<Option<IrStreamEvent>, MapError> {
check_responses_indexes(value)?;
match name {
"response.output_text.delta" => {
if let Some(text) = str_field(value, "delta").filter(|s| !s.is_empty()) {
return Ok(Some(IrStreamEvent::TextDelta { text }));
}
if let Some(content) = logprobs_array(value) {
return Ok(Some(IrStreamEvent::Logprobs {
content,
refusal: None,
}));
}
Ok(None)
}
"response.reasoning_summary_text.delta"
| "response.reasoning.delta"
| "response.reasoning_text.delta" => {
nonempty_delta(value, |text| IrStreamEvent::ReasoningDelta { text })
}
"response.refusal.delta" => {
nonempty_delta(value, |text| IrStreamEvent::RefusalDelta { text })
}
"response.function_call_arguments.delta" => Ok(Some(IrStreamEvent::ToolCallArgDelta {
delta: str_field(value, "delta").unwrap_or_default(),
index: output_index(value),
})),
"response.output_text.annotation.added" => match value.get("annotation") {
Some(annotation) if annotation.is_object() => {
Ok(Some(IrStreamEvent::AnnotationAdded {
annotation: annotation.clone(),
}))
}
_ => Ok(Some(protocol(name, value))),
},
"response.audio.delta" => nonempty_delta(value, |data| IrStreamEvent::AudioDelta { data }),
"response.audio.transcript.delta" => {
nonempty_delta(value, |text| IrStreamEvent::AudioTranscriptDelta { text })
}
"response.custom_tool_call_input.delta" => {
Ok(Some(IrStreamEvent::CustomToolCallInputDelta {
delta: str_field(value, "delta").unwrap_or_default(),
index: output_index(value),
}))
}
"response.content_part.added" => {
let events = content_part_events(value);
if events.len() > 1 {
return Err(MapError::Invalid(
"decode_stream_event cannot represent a Responses content part with more than one event; use decode_stream_events"
.into(),
));
}
Ok(events.into_iter().next())
}
"response.output_item.added" => match item_type(value) {
Some("function_call") => {
let item = value.get("item").unwrap_or(value);
if item
.get("arguments")
.is_some_and(|v| v.is_object() || v.is_array())
{
return Err(super::chat::singular_tool_limit());
}
super::json_text_field(item, "arguments")?;
Ok(Some(IrStreamEvent::ToolCallStart {
id: str_field(item, "call_id")
.or_else(|| str_field(item, "id"))
.unwrap_or_default(),
name: str_field(item, "name").unwrap_or_default(),
thought_signature: None,
index: output_index(value),
}))
}
Some("custom_tool_call") => {
let item = value.get("item").unwrap_or(value);
if item
.get("input")
.is_some_and(|v| v.is_object() || v.is_array())
{
return Err(super::chat::singular_tool_limit());
}
super::json_text_field(item, "input")?;
Ok(Some(IrStreamEvent::CustomToolCallStart {
id: str_field(item, "call_id")
.or_else(|| str_field(item, "id"))
.unwrap_or_default(),
name: str_field(item, "name").unwrap_or_default(),
index: output_index(value),
}))
}
Some("reasoning") => Ok(Some(decode_reasoning_item(name, value))),
Some("message") => {
let item = value.get("item").unwrap_or(value);
let part_count = item
.get("content")
.and_then(Value::as_array)
.map(Vec::len)
.unwrap_or(0);
let events = message_content_events(item);
if part_count > 1 || events.len() > 1 {
return Err(MapError::Invalid(
"decode_stream_event cannot represent every event in this Responses message; use decode_stream_events"
.into(),
));
}
Ok(Some(
events
.into_iter()
.next()
.unwrap_or_else(|| protocol(name, value)),
))
}
_ => {
let item = value.get("item").unwrap_or(value);
let replay = replay_output_item(item);
if replay.len() > 1 {
return Err(MapError::Invalid(
"decode_stream_event cannot represent every event in this Responses item; use decode_stream_events"
.into(),
));
}
Ok(Some(
replay
.into_iter()
.next()
.unwrap_or_else(|| protocol(name, value)),
))
}
},
"response.output_item.done" => match item_type(value) {
Some("function_call") | Some("custom_tool_call") => {
Ok(Some(IrStreamEvent::ToolCallEnd))
}
_ => {
let item = value.get("item").unwrap_or(value);
let replay = replay_output_item(item);
if replay.len() > 1 {
return Err(MapError::Invalid(
"decode_stream_event cannot represent every event in this Responses item; use decode_stream_events"
.into(),
));
}
Ok(Some(
replay
.into_iter()
.next()
.unwrap_or_else(|| protocol(name, value)),
))
}
},
"response.completed" => {
if let Some("cancelled" | "canceled") =
value.pointer("/response/status").and_then(Value::as_str)
{
return Ok(Some(IrStreamEvent::FinishReason {
reason: "cancelled".into(),
vendor: None,
}));
}
if let Some(usage) = value.pointer("/response/usage") {
Ok(Some(usage::from_responses(usage)))
} else {
Ok(Some(IrStreamEvent::Done))
}
}
"response.failed" => {
let last = failure_object(value);
let message = last
.and_then(|err| err.get("message"))
.and_then(Value::as_str)
.filter(|text| !text.is_empty());
let code = last.and_then(last_error_code);
if message.is_some() || code.is_some() {
let detail = match (code, message) {
(Some(code), Some(message)) => format!("{code}: {message}"),
(None, Some(message)) => message.to_string(),
(Some(code), None) => code,
(None, None) => "vendor error".to_string(),
};
return Err(MapError::Invalid(detail));
}
Ok(Some(IrStreamEvent::FinishReason {
reason: "failed".into(),
vendor: None,
}))
}
"response.incomplete" => Ok(Some(IrStreamEvent::FinishReason {
reason: decode_incomplete_reason(value, "incomplete"),
vendor: None,
})),
"response.output_text.done"
| "response.content_part.done"
| "response.function_call_arguments.done"
| "response.refusal.done"
| "response.audio.done"
| "response.audio.transcript.done"
| "response.reasoning_summary_part.added"
| "response.reasoning_summary_part.done"
| "response.reasoning_summary_text.done"
| "response.reasoning_text.done"
| "response.custom_tool_call_input.done" => Ok(None),
"error" => {
let message = str_field(value, "message").filter(|text| !text.is_empty());
let code = str_field(value, "code").filter(|text| !text.is_empty());
let detail = match (code, message) {
(Some(code), Some(message)) => format!("{code}: {message}"),
(None, Some(message)) => message,
(Some(code), None) => code,
(None, None) => "vendor error".to_string(),
};
Err(MapError::Invalid(detail))
}
other => Ok(Some(protocol(other, value))),
}
}
pub(super) fn decode_all(name: &str, value: &Value) -> Result<Vec<IrStreamEvent>, MapError> {
check_responses_indexes(value)?;
if name == "response.created" {
return Ok(response_slot_events(value));
}
if name == "response.output_item.added" {
return added_item_events(value);
}
if name == "response.content_part.added" {
return Ok(content_part_events(value));
}
if name != "response.output_text.delta" {
return Ok(Vec::new());
}
let mut out = Vec::new();
if let Some(text) = str_field(value, "delta").filter(|s| !s.is_empty()) {
out.push(IrStreamEvent::TextDelta { text });
}
if let Some(content) = logprobs_array(value) {
out.push(IrStreamEvent::Logprobs {
content,
refusal: None,
});
}
Ok(out)
}
fn added_item_events(value: &Value) -> Result<Vec<IrStreamEvent>, MapError> {
let item = value.get("item").unwrap_or(value);
let index = output_index(value);
match item_type(value) {
Some("function_call") => {
let mut out = Vec::new();
if let Some(ev) = tool_item_id_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_caller_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_created_by_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_namespace_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_status_event(item, index) {
out.push(ev);
}
out.push(IrStreamEvent::ToolCallStart {
id: str_field(item, "call_id")
.or_else(|| str_field(item, "id"))
.unwrap_or_default(),
name: str_field(item, "name").unwrap_or_default(),
thought_signature: None,
index,
});
if let Some(delta) = super::json_text_field(item, "arguments")? {
out.push(IrStreamEvent::ToolCallArgDelta { delta, index });
}
Ok(out)
}
Some("custom_tool_call") => {
let mut out = Vec::new();
if let Some(ev) = tool_item_id_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_caller_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_created_by_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_namespace_event(item, index) {
out.push(ev);
}
if let Some(ev) = tool_item_status_event(item, index) {
out.push(ev);
}
out.push(IrStreamEvent::CustomToolCallStart {
id: str_field(item, "call_id")
.or_else(|| str_field(item, "id"))
.unwrap_or_default(),
name: str_field(item, "name").unwrap_or_default(),
index,
});
if let Some(delta) = super::json_text_field(item, "input")? {
out.push(IrStreamEvent::CustomToolCallInputDelta { delta, index });
}
Ok(out)
}
Some("message") => Ok(message_content_events(item)),
Some("reasoning") => {
let mut out = Vec::new();
if let Some(signature) = str_field(item, "signature").filter(|s| !s.is_empty()) {
out.push(IrStreamEvent::ReasoningSignature { signature });
} else if let Some(text) = str_field(item, "text").filter(|s| !s.is_empty()) {
out.push(IrStreamEvent::ReasoningDelta { text });
}
let replay = replay_output_item(item);
if replay.is_empty() {
out.extend(reasoning_item_identity_events(item));
}
out.extend(replay);
Ok(out)
}
_ => Ok(replay_output_item(item)),
}
}
pub(super) fn replay_output_item(item: &Value) -> Vec<IrStreamEvent> {
let mut out = Vec::new();
let ty = item.get("type").and_then(Value::as_str).unwrap_or("");
let encrypted = item
.get("encrypted_content")
.and_then(Value::as_str)
.is_some_and(|s| !s.is_empty());
let decomposed = matches!(
ty,
"message" | "function_call" | "custom_tool_call" | "reasoning" | "output_audio"
);
if encrypted || (!ty.is_empty() && !decomposed) {
let item_type = if ty.is_empty() {
"reasoning".to_string()
} else {
ty.to_string()
};
out.push(IrStreamEvent::Protocol {
item_type,
payload: item.clone(),
});
}
out
}
fn content_part_events(value: &Value) -> Vec<IrStreamEvent> {
let part = value.get("part").unwrap_or(value);
part_events(part)
}
fn message_content_events(item: &Value) -> Vec<IrStreamEvent> {
let mut out = message_item_identity_events(item);
let Some(content) = item.get("content").and_then(Value::as_array) else {
return out;
};
out.extend(content.iter().flat_map(part_events));
out
}
pub(super) fn message_item_identity_events(item: &Value) -> Vec<IrStreamEvent> {
let mut out = Vec::new();
if let Some(id) = str_field(item, "id").filter(|id| !id.is_empty()) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_message_id".into(),
payload: json!(id),
});
}
if let Some(status) = str_field(item, "status").filter(|status| !status.is_empty()) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_message_status".into(),
payload: json!(status),
});
}
if let Some(phase) = str_field(item, "phase").filter(|phase| !phase.is_empty()) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_message_phase".into(),
payload: json!(phase),
});
}
if let Some(agent) = item.get("agent").filter(|agent| {
agent
.get("agent_name")
.and_then(Value::as_str)
.is_some_and(|name| !name.is_empty())
}) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_message_agent".into(),
payload: agent.clone(),
});
}
out
}
pub(super) fn tool_item_id_event(item: &Value, index: u32) -> Option<IrStreamEvent> {
let item_id = str_field(item, "id").filter(|id| !id.is_empty())?;
let call_id = str_field(item, "call_id").filter(|id| !id.is_empty())?;
if item_id == call_id {
return None;
}
Some(IrStreamEvent::Protocol {
item_type: "responses_tool_item_id".into(),
payload: json!({ "index": index, "id": item_id }),
})
}
pub(super) fn tool_item_caller_event(item: &Value, index: u32) -> Option<IrStreamEvent> {
let caller = item.get("caller").filter(|caller| {
caller
.get("type")
.and_then(Value::as_str)
.is_some_and(|ty| !ty.is_empty())
})?;
Some(IrStreamEvent::Protocol {
item_type: "responses_tool_caller".into(),
payload: json!({ "index": index, "caller": caller }),
})
}
pub(super) fn tool_item_created_by_event(item: &Value, index: u32) -> Option<IrStreamEvent> {
let created_by = str_field(item, "created_by").filter(|created_by| !created_by.is_empty())?;
Some(IrStreamEvent::Protocol {
item_type: "responses_tool_created_by".into(),
payload: json!({ "index": index, "created_by": created_by }),
})
}
pub(super) fn tool_item_namespace_event(item: &Value, index: u32) -> Option<IrStreamEvent> {
let namespace = str_field(item, "namespace").filter(|namespace| !namespace.is_empty())?;
Some(IrStreamEvent::Protocol {
item_type: "responses_tool_namespace".into(),
payload: json!({ "index": index, "namespace": namespace }),
})
}
pub(super) fn tool_item_status_event(item: &Value, index: u32) -> Option<IrStreamEvent> {
let status = str_field(item, "status").filter(|status| !status.is_empty())?;
Some(IrStreamEvent::Protocol {
item_type: "responses_tool_status".into(),
payload: json!({ "index": index, "status": status }),
})
}
#[derive(Clone, Debug, Default, PartialEq)]
pub(super) struct ResponsesToolExtra {
pub status: Option<String>,
pub namespace: Option<String>,
pub created_by: Option<String>,
pub caller: Option<Value>,
pub item_id: Option<String>,
}
impl ResponsesToolExtra {
pub(super) fn is_empty(&self) -> bool {
self.status.is_none()
&& self.namespace.is_none()
&& self.created_by.is_none()
&& self.caller.is_none()
&& self.item_id.is_none()
}
pub(super) fn merge(&mut self, other: Self) {
if other.status.is_some() {
self.status = other.status;
}
if other.namespace.is_some() {
self.namespace = other.namespace;
}
if other.created_by.is_some() {
self.created_by = other.created_by;
}
if other.caller.is_some() {
self.caller = other.caller;
}
if other.item_id.is_some() {
self.item_id = other.item_id;
}
}
pub(super) fn write_item(&self, item: &mut Value) {
if let Some(item_id) = self.item_id.as_deref().filter(|id| !id.is_empty()) {
item["id"] = json!(item_id);
}
if let Some(status) = self.status.as_deref().filter(|status| !status.is_empty()) {
item["status"] = json!(status);
}
if let Some(namespace) = self.namespace.as_deref().filter(|ns| !ns.is_empty()) {
item["namespace"] = json!(namespace);
}
if let Some(created_by) = self.created_by.as_deref().filter(|by| !by.is_empty()) {
item["created_by"] = json!(created_by);
}
if let Some(caller) = &self.caller {
item["caller"] = caller.clone();
}
}
}
pub(super) fn tool_extra_from_protocol(
item_type: &str,
payload: &Value,
) -> Option<(u32, ResponsesToolExtra)> {
let index = payload
.get("index")
.and_then(Value::as_u64)
.and_then(|index| u32::try_from(index).ok())?;
let mut extra = ResponsesToolExtra::default();
match item_type {
"responses_tool_status" => {
extra.status = nonempty_payload_str(payload, "status");
}
"responses_tool_namespace" => {
extra.namespace = nonempty_payload_str(payload, "namespace");
}
"responses_tool_created_by" => {
extra.created_by = nonempty_payload_str(payload, "created_by");
}
"responses_tool_item_id" => {
extra.item_id = nonempty_payload_str(payload, "id");
}
"responses_tool_caller" => {
extra.caller = payload
.get("caller")
.filter(|caller| {
caller
.get("type")
.and_then(Value::as_str)
.is_some_and(|ty| !ty.is_empty())
})
.cloned();
}
_ => return None,
}
if extra.is_empty() {
None
} else {
Some((index, extra))
}
}
fn nonempty_payload_str(payload: &Value, key: &str) -> Option<String> {
payload
.get(key)
.and_then(Value::as_str)
.filter(|text| !text.is_empty())
.map(str::to_string)
}
pub(super) fn reasoning_item_identity_events(item: &Value) -> Vec<IrStreamEvent> {
if item.get("type").and_then(Value::as_str) != Some("reasoning") {
return Vec::new();
}
let mut out = Vec::new();
if let Some(id) = str_field(item, "id").filter(|id| !id.is_empty()) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_reasoning_id".into(),
payload: json!(id),
});
}
if let Some(status) = str_field(item, "status").filter(|status| !status.is_empty()) {
out.push(IrStreamEvent::Protocol {
item_type: "responses_reasoning_status".into(),
payload: json!(status),
});
}
if let Some(parts) = item.get("content").and_then(Value::as_array) {
let kept: Vec<Value> = parts
.iter()
.filter(|part| {
part.get("type").and_then(Value::as_str) == Some("reasoning_text")
&& str_field(part, "text").is_some_and(|text| !text.is_empty())
})
.cloned()
.collect();
if !kept.is_empty() {
out.push(IrStreamEvent::Protocol {
item_type: "responses_reasoning_content".into(),
payload: json!(kept),
});
}
}
out
}
fn part_events(part: &Value) -> Vec<IrStreamEvent> {
let mut out = Vec::new();
let ty = part.get("type").and_then(Value::as_str);
if matches!(ty, Some("output_text") | Some("text"))
&& let Some(text) = str_field(part, "text").filter(|s| !s.is_empty())
{
out.push(IrStreamEvent::TextDelta { text });
}
if ty == Some("refusal")
&& let Some(text) = str_field(part, "refusal")
.or_else(|| str_field(part, "text"))
.filter(|s| !s.is_empty())
{
out.push(IrStreamEvent::RefusalDelta { text });
}
if let Some(ev) = image_delta_from_output_image(part) {
out.push(ev);
} else if ty == Some("output_image") {
out.push(IrStreamEvent::Protocol {
item_type: "output_image".into(),
payload: part.clone(),
});
}
out.extend(audio_events_from_output_part(part));
out
}
fn audio_events_from_output_part(part: &Value) -> Vec<IrStreamEvent> {
let ty = part.get("type").and_then(Value::as_str);
if !matches!(ty, Some("output_audio") | Some("audio")) {
return Vec::new();
}
let mut out = Vec::new();
if let Some(data) = str_field(part, "data").filter(|s| !s.is_empty()) {
out.push(IrStreamEvent::AudioDelta { data });
}
if let Some(text) = str_field(part, "transcript").filter(|s| !s.is_empty()) {
out.push(IrStreamEvent::AudioTranscriptDelta { text });
}
out
}
pub(super) fn image_delta_from_output_image(part: &Value) -> Option<IrStreamEvent> {
if part.get("type").and_then(Value::as_str) != Some("output_image") {
return None;
}
let url = part.get("image_url").and_then(Value::as_str)?;
let (media_type, data) = url.strip_prefix("data:")?.split_once(";base64,")?;
if data.is_empty() || !media_type.to_ascii_lowercase().starts_with("image/") {
return None;
}
Some(IrStreamEvent::ImageDelta {
media_type: media_type.to_string(),
data: data.to_string(),
})
}
fn response_slot_events(value: &Value) -> Vec<IrStreamEvent> {
let mut out = Vec::new();
if let Some(id) = value
.pointer("/response/id")
.or_else(|| value.get("id"))
.and_then(Value::as_str)
.filter(|text| !text.trim().is_empty())
{
out.push(IrStreamEvent::Protocol {
item_type: "responses_id".into(),
payload: json!(id),
});
}
if let Some(prev) = value
.pointer("/response/previous_response_id")
.or_else(|| value.get("previous_response_id"))
.and_then(Value::as_str)
.filter(|text| !text.trim().is_empty())
{
out.push(IrStreamEvent::Protocol {
item_type: "responses_previous_id".into(),
payload: json!(prev),
});
}
if let Some(ev) = super::complete::created_event(value) {
out.push(ev);
}
if let Some(ev) = super::complete::service_tier_event(value) {
out.push(ev);
}
if let Some(ev) = super::complete::metadata_event(value) {
out.push(ev);
}
if let Some(ev) = super::complete::moderation_event(value) {
out.push(ev);
}
out
}
pub(super) fn logprobs_array(value: &Value) -> Option<Value> {
value
.get("logprobs")
.and_then(Value::as_array)
.filter(|a| !a.is_empty())
.map(|a| Value::Array(a.clone()))
}
pub(super) fn decode_terminal_events(name: &str, value: &Value) -> Option<Vec<IrStreamEvent>> {
if name != "response.completed" && name != "response.incomplete" {
return None;
}
let mut out = Vec::new();
if let Some(items) = value.pointer("/response/output").and_then(Value::as_array) {
for item in items {
let replay = replay_output_item(item);
if replay.is_empty() {
out.extend(reasoning_item_identity_events(item));
}
out.extend(replay);
}
}
if let Some(reason) = terminal_finish_reason(name, value) {
out.push(IrStreamEvent::FinishReason {
reason,
vendor: None,
});
}
if let Some(usage) = value.pointer("/response/usage").filter(|v| v.is_object()) {
out.push(usage::from_responses(usage));
}
let slots = response_slot_events(value);
let had_slots = !slots.is_empty();
out.extend(slots);
(out.len() >= 2 || had_slots).then_some(out)
}
fn terminal_finish_reason(name: &str, value: &Value) -> Option<String> {
match value.pointer("/response/status").and_then(Value::as_str) {
Some("completed") => Some("stop".into()),
Some("incomplete") => Some(decode_incomplete_reason(value, "length")),
Some("failed") => Some("failed".into()),
Some(other) if !other.is_empty() => Some(other.to_string()),
None if name == "response.incomplete" => {
Some(decode_incomplete_reason(value, "incomplete"))
}
None if name == "response.completed"
&& value
.pointer("/response/usage")
.is_some_and(Value::is_object) =>
{
Some("stop".into())
}
_ => None,
}
}
pub(super) fn incomplete_details_reason(ir_reason: &str) -> Option<&'static str> {
match ir_reason {
"content_filter" => Some("content_filter"),
"length" | "max_tokens" => Some("max_output_tokens"),
_ => None,
}
}
fn decode_incomplete_reason(value: &Value, fallback: &str) -> String {
match value
.pointer("/response/incomplete_details/reason")
.and_then(Value::as_str)
{
Some("content_filter") => "content_filter".into(),
Some("max_output_tokens") => "length".into(),
_ => fallback.into(),
}
}
fn output_index(value: &Value) -> u32 {
value
.get("output_index")
.and_then(Value::as_u64)
.and_then(|n| u32::try_from(n).ok())
.unwrap_or(0)
}
fn item_type(value: &Value) -> Option<&str> {
value
.get("item")
.and_then(|item| item.get("type"))
.and_then(Value::as_str)
}
fn decode_reasoning_item(name: &str, value: &Value) -> IrStreamEvent {
let item = value.get("item").unwrap_or(value);
if let Some(signature) = str_field(item, "signature").filter(|s| !s.is_empty()) {
return IrStreamEvent::ReasoningSignature { signature };
}
if let Some(text) = str_field(item, "text").filter(|s| !s.is_empty()) {
return IrStreamEvent::ReasoningDelta { text };
}
protocol(name, value)
}
fn nonempty_delta(
value: &Value,
wrap: impl FnOnce(String) -> IrStreamEvent,
) -> Result<Option<IrStreamEvent>, MapError> {
Ok(str_field(value, "delta")
.filter(|s| !s.is_empty())
.map(wrap))
}
fn check_responses_indexes(value: &Value) -> Result<(), MapError> {
check_index(value, "output_index", MAX_TOOL_CALL_INDEX, "output")?;
check_index(value, "content_index", MAX_TOOL_CALL_INDEX, "content")?;
Ok(())
}
pub(super) fn annotation_without_segment_text(annotation: &Value) -> Value {
let mut annotation = annotation.clone();
if let Some(obj) = annotation.as_object_mut() {
obj.remove("segment_text");
}
annotation
}
pub(super) fn encode(ev: &IrStreamEvent) -> Result<RawSse, MapError> {
let (event, data) = match ev {
IrStreamEvent::TextDelta { text } => (
"response.output_text.delta",
json!({
"type": "response.output_text.delta",
"output_index": 0,
"delta": text
}),
),
IrStreamEvent::Logprobs { content, refusal } => {
let merged = super::chat::merged_logprob_items(content, refusal);
let mut data = json!({
"type": "response.output_text.delta",
"output_index": 0,
"delta": "",
});
if !merged.is_empty() {
data["logprobs"] = Value::Array(merged);
}
("response.output_text.delta", data)
}
IrStreamEvent::RefusalDelta { text } => (
"response.refusal.delta",
json!({
"type": "response.refusal.delta",
"output_index": 0,
"delta": text
}),
),
IrStreamEvent::ReasoningDelta { text } => (
"response.output_item.added",
json!({
"type": "response.output_item.added",
"output_index": 0,
"item": { "type": "reasoning", "text": text }
}),
),
IrStreamEvent::ReasoningSignature { signature } => (
"response.output_item.added",
json!({
"type": "response.output_item.added",
"output_index": 0,
"item": { "type": "reasoning", "signature": signature }
}),
),
IrStreamEvent::ToolCallStart {
id, name, index, ..
} => (
"response.output_item.added",
json!({
"type": "response.output_item.added",
"output_index": index,
"item": {
"type": "function_call",
"id": id,
"call_id": id,
"name": name,
"arguments": ""
}
}),
),
IrStreamEvent::AnnotationAdded { annotation } => {
let annotation = annotation_without_segment_text(annotation);
(
"response.output_text.annotation.added",
json!({
"type": "response.output_text.annotation.added",
"output_index": 0,
"content_index": 0,
"annotation_index": 0,
"annotation": annotation
}),
)
}
IrStreamEvent::AudioDelta { data } => (
"response.audio.delta",
json!({
"type": "response.audio.delta",
"delta": data
}),
),
IrStreamEvent::ImageDelta { media_type, data } => (
"response.output_item.added",
json!({
"type": "response.output_item.added",
"output_index": 0,
"item": {
"type": "message",
"role": "assistant",
"content": [{
"type": "output_image",
"image_url": format!("data:{media_type};base64,{data}")
}]
}
}),
),
IrStreamEvent::AudioTranscriptDelta { text } => (
"response.audio.transcript.delta",
json!({
"type": "response.audio.transcript.delta",
"delta": text
}),
),
IrStreamEvent::CustomToolCallStart { id, name, index } => (
"response.output_item.added",
json!({
"type": "response.output_item.added",
"output_index": index,
"item": {
"type": "custom_tool_call",
"id": id,
"call_id": id,
"name": name,
"input": ""
}
}),
),
IrStreamEvent::CustomToolCallInputDelta { delta, index } => (
"response.custom_tool_call_input.delta",
json!({
"type": "response.custom_tool_call_input.delta",
"output_index": index,
"delta": delta
}),
),
IrStreamEvent::ToolCallArgDelta { delta, index } => (
"response.function_call_arguments.delta",
json!({
"type": "response.function_call_arguments.delta",
"output_index": index,
"delta": delta
}),
),
IrStreamEvent::ToolCallEnd => (
"response.output_item.done",
json!({
"type": "response.output_item.done",
"output_index": 0,
"item": { "type": "function_call" }
}),
),
IrStreamEvent::Usage {
prompt_tokens,
completion_tokens,
cache_read_tokens,
cache_write_tokens,
reasoning_tokens,
..
} => (
"response.completed",
usage::encode_responses(
*prompt_tokens,
*completion_tokens,
*cache_read_tokens,
*cache_write_tokens,
*reasoning_tokens,
),
),
IrStreamEvent::FinishReason { reason, .. } => {
let (event, status) = match reason.as_str() {
"failed" => ("response.failed", "failed"),
"incomplete" | "length" | "max_tokens" | "content_filter" => {
("response.incomplete", "incomplete")
}
"stop" => ("response.completed", "completed"),
other => ("response.completed", other),
};
let mut response = json!({ "status": status });
if let Some(detail) = incomplete_details_reason(reason) {
response["incomplete_details"] = json!({ "reason": detail });
}
(
event,
json!({
"type": event,
"response": response
}),
)
}
IrStreamEvent::Created { unix } => (
"response.created",
json!({
"type": "response.created",
"response": { "created_at": unix, "status": "in_progress" }
}),
),
IrStreamEvent::ServiceTier { tier } => (
"response.created",
json!({
"type": "response.created",
"response": { "service_tier": tier, "status": "in_progress" }
}),
),
IrStreamEvent::Metadata { metadata } => (
"response.created",
json!({
"type": "response.created",
"response": { "metadata": metadata, "status": "in_progress" }
}),
),
IrStreamEvent::Moderation { input, output } => {
let mut response = json!({ "status": "completed" });
if let Some(value) = super::complete::responses_moderation_value(input, output) {
response["moderation"] = value;
}
(
"response.completed",
json!({
"type": "response.completed",
"response": response
}),
)
}
IrStreamEvent::Done => (
"response.completed",
json!({
"type": "response.completed",
"response": { "status": "completed" }
}),
),
IrStreamEvent::Diagnostics { .. }
| IrStreamEvent::Container { .. }
| IrStreamEvent::ContextManagement { .. }
| IrStreamEvent::StopSequence { .. }
| IrStreamEvent::SearchEntryPoint { .. } => (
"response.created",
json!({
"type": "response.created",
"response": { "status": "in_progress" }
}),
),
IrStreamEvent::Protocol { .. } | IrStreamEvent::Unknown { .. } => {
return Err(MapError::Invalid(
"protocol/unknown events are encoded at the stream root".into(),
));
}
};
Ok(RawSse {
event: Some(event.into()),
data: data.to_string(),
})
}