use router_protocol::{CallId, CanonicalEvent, ItemId, OutputItemKind, ResponseId};
use serde_json::{Value, json};
use std::collections::HashMap;
use thiserror::Error;
#[derive(Debug, Error, PartialEq, Eq)]
pub enum CodexSseEncodeError {
#[error("canonical event sequence is invalid for Codex Responses SSE")]
InvalidSequence,
#[error("canonical event data could not be encoded for Codex Responses SSE")]
InvalidData,
}
#[derive(Clone, Debug)]
enum ActiveItem {
Message {
id: ItemId,
text: String,
done: bool,
},
FunctionCall {
id: ItemId,
call_id: CallId,
name: String,
namespace: Option<String>,
arguments: String,
done: bool,
},
}
impl ActiveItem {
fn wire_value(&self, status: &'static str) -> Value {
match self {
Self::Message { id, text, .. } => json!({
"id": id.as_str(),
"type": "message",
"status": status,
"role": "assistant",
"content": if text.is_empty() {
Vec::<Value>::new()
} else {
vec![json!({
"type": "output_text",
"text": text,
"annotations": []
})]
}
}),
Self::FunctionCall {
id,
call_id,
name,
namespace,
arguments,
..
} => {
let mut item = json!({
"id": id.as_str(),
"type": "function_call",
"status": status,
"call_id": call_id.as_str(),
"name": name,
"arguments": arguments
});
if let Some(namespace) = namespace {
item["namespace"] = Value::String(namespace.clone());
}
item
}
}
}
fn is_done(&self) -> bool {
match self {
Self::Message { done, .. } | Self::FunctionCall { done, .. } => *done,
}
}
}
pub struct CodexSseEncoder {
model: String,
response_id: Option<ResponseId>,
items: HashMap<ItemId, ActiveItem>,
item_order: Vec<ItemId>,
}
impl CodexSseEncoder {
pub fn new(model: impl Into<String>) -> Self {
Self {
model: model.into(),
response_id: None,
items: HashMap::new(),
item_order: Vec::new(),
}
}
pub fn encode(&mut self, event: &CanonicalEvent) -> Result<Value, CodexSseEncodeError> {
match event {
CanonicalEvent::ResponseCreated { response_id, .. } => {
if self.response_id.replace(response_id.clone()).is_some() {
return Err(CodexSseEncodeError::InvalidSequence);
}
Ok(json!({
"type": "response.created",
"sequence_number": event.sequence(),
"response": self.response_value(response_id, "in_progress", None)
}))
}
CanonicalEvent::ResponseInProgress {
response_id,
sequence,
} => {
self.ensure_response(response_id)?;
Ok(json!({
"type": "response.in_progress",
"sequence_number": sequence,
"response": self.response_value(response_id, "in_progress", None)
}))
}
CanonicalEvent::OutputItemAdded {
response_id,
item_id,
sequence,
item_kind,
call_id,
name,
namespace,
} => {
self.ensure_response(response_id)?;
if self.items.contains_key(item_id) {
return Err(CodexSseEncodeError::InvalidSequence);
}
let item = match item_kind {
OutputItemKind::Message
if call_id.is_none() && name.is_none() && namespace.is_none() =>
{
ActiveItem::Message {
id: item_id.clone(),
text: String::new(),
done: false,
}
}
OutputItemKind::FunctionCall => ActiveItem::FunctionCall {
id: item_id.clone(),
call_id: call_id.clone().ok_or(CodexSseEncodeError::InvalidData)?,
name: name.clone().ok_or(CodexSseEncodeError::InvalidData)?,
namespace: namespace.clone(),
arguments: String::new(),
done: false,
},
OutputItemKind::Message => return Err(CodexSseEncodeError::InvalidData),
};
let output_index = self.item_order.len();
let wire_item = item.wire_value("in_progress");
self.item_order.push(item_id.clone());
self.items.insert(item_id.clone(), item);
Ok(json!({
"type": "response.output_item.added",
"sequence_number": sequence,
"output_index": output_index,
"item": wire_item
}))
}
CanonicalEvent::ContentPartAdded {
response_id,
item_id,
sequence,
} => {
self.ensure_message(response_id, item_id)?;
Ok(json!({
"type": "response.content_part.added",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": self.output_index(item_id)?,
"content_index": 0,
"part": {"type": "output_text", "text": "", "annotations": []}
}))
}
CanonicalEvent::OutputTextDelta {
response_id,
item_id,
sequence,
delta,
} => {
let output_index = self.output_index(item_id)?;
let item = self.ensure_message_mut(response_id, item_id)?;
let ActiveItem::Message { text, done, .. } = item else {
return Err(CodexSseEncodeError::InvalidSequence);
};
if *done {
return Err(CodexSseEncodeError::InvalidSequence);
}
text.push_str(delta);
Ok(json!({
"type": "response.output_text.delta",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": output_index,
"content_index": 0,
"delta": delta
}))
}
CanonicalEvent::OutputTextDone {
response_id,
item_id,
sequence,
text,
} => {
let output_index = self.output_index(item_id)?;
let item = self.ensure_message_mut(response_id, item_id)?;
let ActiveItem::Message {
text: accumulated,
done,
..
} = item
else {
return Err(CodexSseEncodeError::InvalidSequence);
};
if *done {
return Err(CodexSseEncodeError::InvalidSequence);
}
accumulated.clone_from(text);
Ok(json!({
"type": "response.output_text.done",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": output_index,
"content_index": 0,
"text": text
}))
}
CanonicalEvent::ContentPartDone {
response_id,
item_id,
sequence,
} => {
let output_index = self.output_index(item_id)?;
let item = self.ensure_message(response_id, item_id)?;
let ActiveItem::Message { text, .. } = item else {
return Err(CodexSseEncodeError::InvalidSequence);
};
Ok(json!({
"type": "response.content_part.done",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": output_index,
"content_index": 0,
"part": {"type": "output_text", "text": text, "annotations": []}
}))
}
CanonicalEvent::FunctionCallArgumentsDelta {
response_id,
item_id,
call_id,
sequence,
delta,
} => {
let output_index = self.output_index(item_id)?;
let item = self.ensure_function_mut(response_id, item_id, call_id)?;
let ActiveItem::FunctionCall {
arguments, done, ..
} = item
else {
return Err(CodexSseEncodeError::InvalidSequence);
};
if *done {
return Err(CodexSseEncodeError::InvalidSequence);
}
arguments.push_str(delta);
Ok(json!({
"type": "response.function_call_arguments.delta",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": output_index,
"delta": delta
}))
}
CanonicalEvent::FunctionCallArgumentsDone {
response_id,
item_id,
call_id,
sequence,
arguments,
} => {
let output_index = self.output_index(item_id)?;
let encoded_arguments = serde_json::to_string(arguments)
.map_err(|_| CodexSseEncodeError::InvalidData)?;
let item = self.ensure_function_mut(response_id, item_id, call_id)?;
let ActiveItem::FunctionCall {
arguments, done, ..
} = item
else {
return Err(CodexSseEncodeError::InvalidSequence);
};
if *done {
return Err(CodexSseEncodeError::InvalidSequence);
}
arguments.clone_from(&encoded_arguments);
Ok(json!({
"type": "response.function_call_arguments.done",
"sequence_number": sequence,
"item_id": item_id.as_str(),
"output_index": output_index,
"arguments": encoded_arguments
}))
}
CanonicalEvent::OutputItemDone {
response_id,
item_id,
sequence,
call_id,
} => {
self.ensure_response(response_id)?;
let output_index = self.output_index(item_id)?;
let item = self
.items
.get_mut(item_id)
.ok_or(CodexSseEncodeError::InvalidSequence)?;
match item {
ActiveItem::Message { done, .. } if call_id.is_none() => *done = true,
ActiveItem::FunctionCall {
call_id: expected,
done,
..
} if call_id.as_ref() == Some(expected) => *done = true,
ActiveItem::Message { .. } | ActiveItem::FunctionCall { .. } => {
return Err(CodexSseEncodeError::InvalidSequence);
}
}
Ok(json!({
"type": "response.output_item.done",
"sequence_number": sequence,
"output_index": output_index,
"item": item.wire_value("completed")
}))
}
CanonicalEvent::ResponseCompleted {
response_id,
sequence,
usage,
} => {
self.ensure_response(response_id)?;
if self.items.values().any(|item| !item.is_done()) {
return Err(CodexSseEncodeError::InvalidSequence);
}
let total_tokens = usage
.input_tokens
.checked_add(usage.output_tokens)
.ok_or(CodexSseEncodeError::InvalidData)?;
let usage = json!({
"input_tokens": usage.input_tokens,
"input_tokens_details": {
"cached_tokens": 0,
"cache_write_tokens": 0
},
"output_tokens": usage.output_tokens,
"output_tokens_details": {"reasoning_tokens": 0},
"total_tokens": total_tokens
});
Ok(json!({
"type": "response.completed",
"sequence_number": sequence,
"response": self.response_value(response_id, "completed", Some(usage))
}))
}
CanonicalEvent::ResponseFailed {
response_id,
sequence,
error,
} => {
self.ensure_response(response_id)?;
Ok(json!({
"type": "response.failed",
"sequence_number": sequence,
"response": {
"id": response_id.as_str(),
"object": "response",
"status": "failed",
"error": {
"type": "server_error",
"code": error.code.as_str(),
"message": error.message,
"details": error.details
}
}
}))
}
CanonicalEvent::Error { sequence, error } => Ok(json!({
"type": "error",
"sequence_number": sequence,
"error": {
"type": "server_error",
"code": error.code.as_str(),
"message": error.message,
"details": error.details
}
})),
}
}
fn response_value(
&self,
response_id: &ResponseId,
status: &str,
usage: Option<Value>,
) -> Value {
let output = self
.item_order
.iter()
.filter_map(|item_id| self.items.get(item_id))
.map(|item| {
item.wire_value(if item.is_done() {
"completed"
} else {
"in_progress"
})
})
.collect::<Vec<_>>();
json!({
"id": response_id.as_str(),
"object": "response",
"created_at": 0,
"status": status,
"model": self.model,
"output": output,
"usage": usage,
"error": null,
"incomplete_details": null
})
}
fn ensure_response(&self, response_id: &ResponseId) -> Result<(), CodexSseEncodeError> {
if self.response_id.as_ref() == Some(response_id) {
Ok(())
} else {
Err(CodexSseEncodeError::InvalidSequence)
}
}
fn output_index(&self, item_id: &ItemId) -> Result<usize, CodexSseEncodeError> {
self.item_order
.iter()
.position(|candidate| candidate == item_id)
.ok_or(CodexSseEncodeError::InvalidSequence)
}
fn ensure_message(
&self,
response_id: &ResponseId,
item_id: &ItemId,
) -> Result<&ActiveItem, CodexSseEncodeError> {
self.ensure_response(response_id)?;
match self.items.get(item_id) {
Some(item @ ActiveItem::Message { .. }) => Ok(item),
Some(ActiveItem::FunctionCall { .. }) | None => {
Err(CodexSseEncodeError::InvalidSequence)
}
}
}
fn ensure_message_mut(
&mut self,
response_id: &ResponseId,
item_id: &ItemId,
) -> Result<&mut ActiveItem, CodexSseEncodeError> {
self.ensure_response(response_id)?;
match self.items.get_mut(item_id) {
Some(item @ ActiveItem::Message { .. }) => Ok(item),
Some(ActiveItem::FunctionCall { .. }) | None => {
Err(CodexSseEncodeError::InvalidSequence)
}
}
}
fn ensure_function_mut(
&mut self,
response_id: &ResponseId,
item_id: &ItemId,
call_id: &CallId,
) -> Result<&mut ActiveItem, CodexSseEncodeError> {
self.ensure_response(response_id)?;
let item = self
.items
.get_mut(item_id)
.ok_or(CodexSseEncodeError::InvalidSequence)?;
if matches!(
&*item,
ActiveItem::FunctionCall {
call_id: expected,
..
} if expected == call_id
) {
Ok(item)
} else {
Err(CodexSseEncodeError::InvalidSequence)
}
}
}
trait CanonicalEventSequence {
fn sequence(&self) -> u64;
}
impl CanonicalEventSequence for CanonicalEvent {
fn sequence(&self) -> u64 {
match self {
Self::ResponseCreated { sequence, .. }
| Self::ResponseInProgress { sequence, .. }
| Self::OutputItemAdded { sequence, .. }
| Self::OutputItemDone { sequence, .. }
| Self::ContentPartAdded { sequence, .. }
| Self::ContentPartDone { sequence, .. }
| Self::OutputTextDelta { sequence, .. }
| Self::OutputTextDone { sequence, .. }
| Self::FunctionCallArgumentsDelta { sequence, .. }
| Self::FunctionCallArgumentsDone { sequence, .. }
| Self::ResponseCompleted { sequence, .. }
| Self::ResponseFailed { sequence, .. }
| Self::Error { sequence, .. } => *sequence,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use router_protocol::{
InterruptionCause, PublicError, PublicErrorCode, RequestId, SafeResumeBoundary, Usage,
};
fn response_id() -> ResponseId {
ResponseId::parse("resp-sse").expect("response ID")
}
fn item_id(value: &str) -> ItemId {
ItemId::parse(value).expect("item ID")
}
fn call_id() -> CallId {
CallId::parse("call-sse").expect("call ID")
}
fn event_types(events: &[Value]) -> Vec<&str> {
events
.iter()
.map(|event| event["type"].as_str().expect("event type"))
.collect()
}
#[test]
fn encodes_exact_message_and_namespaced_function_lifecycle() {
let response_id = response_id();
let message_id = item_id("msg-sse");
let function_id = item_id("fc-sse");
let call_id = call_id();
let canonical = vec![
CanonicalEvent::ResponseCreated {
response_id: response_id.clone(),
request_id: RequestId::parse("request-sse").expect("request ID"),
sequence: 0,
},
CanonicalEvent::ResponseInProgress {
response_id: response_id.clone(),
sequence: 1,
},
CanonicalEvent::OutputItemAdded {
response_id: response_id.clone(),
item_id: message_id.clone(),
sequence: 2,
item_kind: OutputItemKind::Message,
call_id: None,
name: None,
namespace: None,
},
CanonicalEvent::ContentPartAdded {
response_id: response_id.clone(),
item_id: message_id.clone(),
sequence: 3,
},
CanonicalEvent::OutputTextDelta {
response_id: response_id.clone(),
item_id: message_id.clone(),
sequence: 4,
delta: "Checking. ".to_owned(),
},
CanonicalEvent::OutputTextDone {
response_id: response_id.clone(),
item_id: message_id.clone(),
sequence: 5,
text: "Checking. ".to_owned(),
},
CanonicalEvent::ContentPartDone {
response_id: response_id.clone(),
item_id: message_id.clone(),
sequence: 6,
},
CanonicalEvent::OutputItemDone {
response_id: response_id.clone(),
item_id: message_id,
sequence: 7,
call_id: None,
},
CanonicalEvent::OutputItemAdded {
response_id: response_id.clone(),
item_id: function_id.clone(),
sequence: 8,
item_kind: OutputItemKind::FunctionCall,
call_id: Some(call_id.clone()),
name: Some("spawn_agent".to_owned()),
namespace: Some("collaboration".to_owned()),
},
CanonicalEvent::FunctionCallArgumentsDelta {
response_id: response_id.clone(),
item_id: function_id.clone(),
call_id: call_id.clone(),
sequence: 9,
delta: "{\"task\":".to_owned(),
},
CanonicalEvent::FunctionCallArgumentsDelta {
response_id: response_id.clone(),
item_id: function_id.clone(),
call_id: call_id.clone(),
sequence: 10,
delta: "\"fixture\"}".to_owned(),
},
CanonicalEvent::FunctionCallArgumentsDone {
response_id: response_id.clone(),
item_id: function_id.clone(),
call_id: call_id.clone(),
sequence: 11,
arguments: json!({"task": "fixture"}),
},
CanonicalEvent::OutputItemDone {
response_id: response_id.clone(),
item_id: function_id,
sequence: 12,
call_id: Some(call_id),
},
CanonicalEvent::ResponseCompleted {
response_id,
sequence: 13,
usage: Usage {
input_tokens: 17,
output_tokens: 11,
},
},
];
let mut encoder = CodexSseEncoder::new("router-auto");
let encoded = canonical
.iter()
.map(|event| encoder.encode(event).expect("valid event sequence"))
.collect::<Vec<_>>();
assert_eq!(
event_types(&encoded),
[
"response.created",
"response.in_progress",
"response.output_item.added",
"response.content_part.added",
"response.output_text.delta",
"response.output_text.done",
"response.content_part.done",
"response.output_item.done",
"response.output_item.added",
"response.function_call_arguments.delta",
"response.function_call_arguments.delta",
"response.function_call_arguments.done",
"response.output_item.done",
"response.completed",
]
);
assert_eq!(
encoded[2]
.as_object()
.expect("added event object")
.keys()
.map(String::as_str)
.collect::<Vec<_>>(),
["item", "output_index", "sequence_number", "type"]
);
assert_eq!(encoded[2]["item"]["type"], "message");
assert_eq!(encoded[2]["item"]["content"], json!([]));
assert_eq!(encoded[4]["item_id"], "msg-sse");
assert_eq!(encoded[7]["item"]["content"][0]["text"], "Checking. ");
assert_eq!(encoded[8]["item"]["type"], "function_call");
assert_eq!(encoded[8]["item"]["namespace"], "collaboration");
assert_eq!(encoded[8]["item"]["name"], "spawn_agent");
assert_eq!(encoded[12]["item"]["arguments"], "{\"task\":\"fixture\"}");
assert_eq!(encoded[13]["response"]["status"], "completed");
assert_eq!(encoded[13]["response"]["usage"]["total_tokens"], 28);
assert_eq!(
encoded[13]["response"]["output"].as_array().map(Vec::len),
Some(2)
);
let rendered = serde_json::to_string(&encoded).expect("render encoded events");
assert!(!rendered.contains("encrypted_content"));
assert!(!rendered.contains("reasoning_summary"));
}
#[test]
fn rejects_text_delta_without_an_active_message() {
let response_id = response_id();
let mut encoder = CodexSseEncoder::new("router-auto");
encoder
.encode(&CanonicalEvent::ResponseCreated {
response_id: response_id.clone(),
request_id: RequestId::parse("request-sse").expect("request ID"),
sequence: 0,
})
.expect("created event");
assert_eq!(
encoder.encode(&CanonicalEvent::OutputTextDelta {
response_id,
item_id: item_id("missing-item"),
sequence: 1,
delta: "must fail".to_owned(),
}),
Err(CodexSseEncodeError::InvalidSequence)
);
}
#[test]
fn rejects_terminal_usage_total_overflow() {
let response_id = response_id();
let mut encoder = CodexSseEncoder::new("router-auto");
encoder
.encode(&CanonicalEvent::ResponseCreated {
response_id: response_id.clone(),
request_id: RequestId::parse("request-sse").expect("request ID"),
sequence: 0,
})
.expect("created event");
assert_eq!(
encoder.encode(&CanonicalEvent::ResponseCompleted {
response_id,
sequence: 1,
usage: Usage {
input_tokens: u64::MAX,
output_tokens: 1,
},
}),
Err(CodexSseEncodeError::InvalidData)
);
}
#[test]
fn response_failure_is_terminal_and_redacted() {
let response_id = response_id();
let mut encoder = CodexSseEncoder::new("router-auto");
encoder
.encode(&CanonicalEvent::ResponseCreated {
response_id: response_id.clone(),
request_id: RequestId::parse("request-sse").expect("request ID"),
sequence: 0,
})
.expect("created event");
let error = PublicError::redacted(
PublicErrorCode::ProviderFailure,
RequestId::parse("request-sse").expect("request ID"),
false,
);
let event = encoder
.encode(&CanonicalEvent::ResponseFailed {
response_id,
sequence: 1,
error,
})
.expect("failed response");
assert_eq!(event["type"], "response.failed");
assert_eq!(event["response"]["status"], "failed");
assert_eq!(event["response"]["error"]["code"], "provider_failure");
assert_eq!(event["response"].get("output"), None);
}
#[test]
fn response_failure_preserves_fixed_safe_resume_details() {
let response_id = response_id();
let mut encoder = CodexSseEncoder::new("router-auto");
encoder
.encode(&CanonicalEvent::ResponseCreated {
response_id: response_id.clone(),
request_id: RequestId::parse("request-resume").expect("request ID"),
sequence: 0,
})
.expect("created event");
let event = encoder
.encode(&CanonicalEvent::ResponseFailed {
response_id,
sequence: 1,
error: PublicError::interrupted(
RequestId::parse("request-resume").expect("request ID"),
InterruptionCause::Timeout,
SafeResumeBoundary::ToolCall,
),
})
.expect("failed response");
let interruption = &event["response"]["error"]["details"]["interruption"];
assert_eq!(interruption["cause"], "timeout");
assert_eq!(interruption["boundary"], "tool_call");
assert_eq!(interruption["automatic_replay"], false);
assert_eq!(interruption["safe_resume"]["action"], "start_new_turn");
assert_eq!(
interruption["safe_resume"]["repeat_completed_tool_calls"],
false
);
}
}