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 decode(name: &str, value: &Value) -> Result<Option<IrStreamEvent>, MapError> {
check_responses_indexes(value)?;
match name {
"response.output_text.delta" => {
nonempty_delta(value, |text| IrStreamEvent::TextDelta { text })
}
"response.reasoning_summary_text.delta" | "response.reasoning.delta" => {
nonempty_delta(value, |text| IrStreamEvent::ReasoningDelta { text })
}
"response.refusal.delta" => nonempty_delta(value, |text| IrStreamEvent::TextDelta { text }),
"response.function_call_arguments.delta" => Ok(Some(IrStreamEvent::ToolCallArgDelta {
delta: str_field(value, "delta").unwrap_or_default(),
index: output_index(value),
})),
"response.output_item.added" => match item_type(value) {
Some("function_call") => {
let item = value.get("item").unwrap_or(value);
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("reasoning") => Ok(Some(decode_reasoning_item(name, value))),
_ => Ok(Some(protocol(name, value))),
},
"response.output_item.done" => match item_type(value) {
Some("function_call") => Ok(Some(IrStreamEvent::ToolCallEnd)),
_ => Ok(Some(protocol(name, value))),
},
"response.completed" => {
if let Some(usage) = value.pointer("/response/usage") {
Ok(Some(usage::from_responses(usage)))
} else {
Ok(Some(IrStreamEvent::Done))
}
}
"response.failed" => Ok(Some(IrStreamEvent::FinishReason {
reason: "failed".into(),
})),
"response.incomplete" => Ok(Some(IrStreamEvent::FinishReason {
reason: "incomplete".into(),
})),
other => Ok(Some(protocol(other, value))),
}
}
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 {
if item
.get("encrypted_content")
.and_then(Value::as_str)
.is_none_or(|s| s.is_empty())
{
continue;
}
out.push(IrStreamEvent::Protocol {
item_type: item
.get("type")
.and_then(Value::as_str)
.unwrap_or("reasoning")
.to_string(),
payload: item.clone(),
});
}
}
if let Some(reason) = terminal_finish_reason(name, value) {
out.push(IrStreamEvent::FinishReason { reason });
}
if let Some(usage) = value.pointer("/response/usage").filter(|v| v.is_object()) {
out.push(usage::from_responses(usage));
}
(out.len() >= 2).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("length".into()),
Some("failed") => Some("failed".into()),
Some(other) if !other.is_empty() => Some(other.to_string()),
None if name == "response.incomplete" => Some("incomplete".into()),
None if name == "response.completed"
&& value
.pointer("/response/usage")
.is_some_and(Value::is_object) =>
{
Some("stop".into())
}
_ => None,
}
}
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 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::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::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,
reasoning_tokens,
..
} => (
"response.completed",
usage::encode_responses(
*prompt_tokens,
*completion_tokens,
*cache_read_tokens,
*reasoning_tokens,
),
),
IrStreamEvent::FinishReason { reason } => {
let (event, status) = match reason.as_str() {
"failed" => ("response.failed", "failed"),
"incomplete" | "length" | "max_tokens" => ("response.incomplete", "incomplete"),
"stop" => ("response.completed", "completed"),
other => ("response.completed", other),
};
(
event,
json!({
"type": event,
"response": { "status": status }
}),
)
}
IrStreamEvent::Done => (
"response.completed",
json!({
"type": "response.completed",
"response": { "status": "completed" }
}),
),
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(),
})
}