use std::collections::{HashMap, HashSet};
use futures_util::StreamExt;
use serde::Deserialize;
use tokio::sync::mpsc;
use crate::{
ContentBlockDelta, ContentBlockStart, HostedToolSearchCall, HostedWebSearchCall,
ImageGenerationCall, ImageGenerationResult, ProviderError, ProviderEvent, ProviderEventStream,
ProviderId, ReasoningFormat, ReasoningProvenance, ResponseHeaders, Role, TokenUsage,
WebSearchAction,
};
use super::model::encode_responses_tool_use_id;
pub fn spawn_event_stream(response: reqwest::Response) -> ProviderEventStream {
spawn_event_stream_with_provenance(response, ProviderId::new("openai"), String::new())
}
pub(crate) fn spawn_event_stream_with_provenance(
response: reqwest::Response,
provider: ProviderId,
requested_model: String,
) -> ProviderEventStream {
let (tx, rx) = mpsc::unbounded_channel();
tokio::spawn(async move {
if let Err(error) = forward_events(response, tx.clone(), provider, requested_model).await {
let _ = tx.send(Err(error));
}
});
rx
}
async fn forward_events(
response: reqwest::Response,
tx: mpsc::UnboundedSender<Result<ProviderEvent, ProviderError>>,
provider: ProviderId,
requested_model: String,
) -> Result<(), ProviderError> {
let receiver_closed = response_headers_event(response.headers())
.map(|headers| tx.send(Ok(headers)).is_err())
.unwrap_or(false);
if receiver_closed {
return Ok(());
}
let mut bytes_stream = response.bytes_stream();
let mut buffer = Vec::new();
let mut state = StreamState::new(provider, requested_model);
while let Some(chunk) = bytes_stream.next().await {
let chunk = chunk.map_err(ProviderError::Transport)?;
buffer.extend_from_slice(&chunk);
while let Some((frame_end, delimiter_len)) = find_frame_boundary(&buffer) {
let frame = buffer.drain(..frame_end).collect::<Vec<_>>();
buffer.drain(..delimiter_len);
for event in parse_frame(&frame, &mut state)? {
if tx.send(Ok(event)).is_err() {
return Ok(());
}
}
}
}
if !buffer.is_empty() {
for event in parse_frame(&buffer, &mut state)? {
let _ = tx.send(Ok(event));
}
}
Ok(())
}
pub(crate) fn response_headers_event(headers: &http::HeaderMap) -> Option<ProviderEvent> {
let values = headers
.iter()
.filter_map(|(name, value)| {
value
.to_str()
.ok()
.map(|value| (name.as_str().to_string(), value.to_string()))
})
.collect::<Vec<_>>();
(!values.is_empty()).then_some(ProviderEvent::ResponseHeaders(ResponseHeaders { values }))
}
pub(crate) struct StreamState {
ignored_output_indices: HashSet<usize>,
text_delta_seen: HashSet<usize>,
function_delta_seen: HashSet<usize>,
tool_search_delta_seen: HashSet<usize>,
reasoning_delta_seen: HashSet<usize>,
reasoning_encrypted_content_seen: HashSet<usize>,
reasoning_output_indices: HashMap<String, usize>,
reasoning_provenance: ReasoningProvenance,
}
impl Default for StreamState {
fn default() -> Self {
Self::new(ProviderId::new("openai"), String::new())
}
}
impl StreamState {
pub(crate) fn new(provider: ProviderId, requested_model: String) -> Self {
Self {
ignored_output_indices: HashSet::new(),
text_delta_seen: HashSet::new(),
function_delta_seen: HashSet::new(),
tool_search_delta_seen: HashSet::new(),
reasoning_delta_seen: HashSet::new(),
reasoning_encrypted_content_seen: HashSet::new(),
reasoning_output_indices: HashMap::new(),
reasoning_provenance: ReasoningProvenance {
provider,
model: requested_model,
format: ReasoningFormat::OpenAiEncrypted,
},
}
}
fn backfill_reasoning_encrypted_content(
&mut self,
output: &[ResponsesOutputItem],
) -> Vec<ProviderEvent> {
let mut events = Vec::new();
for item in output {
let Some((id, encrypted_content)) = item.reasoning_encrypted_content() else {
continue;
};
let Some(index) = self.reasoning_output_indices.get(id).copied() else {
continue;
};
if !self.reasoning_encrypted_content_seen.insert(index) {
continue;
}
events.push(ProviderEvent::ContentBlockDelta {
index,
delta: ContentBlockDelta::ThinkingEncryptedContent(encrypted_content.to_string()),
});
}
events
}
}
fn parse_frame(frame: &[u8], state: &mut StreamState) -> Result<Vec<ProviderEvent>, ProviderError> {
let frame = std::str::from_utf8(frame)
.map_err(|error| ProviderError::MalformedStream(error.to_string()))?;
let mut data_lines = Vec::new();
for raw_line in frame.lines() {
let line = raw_line.strip_suffix('\r').unwrap_or(raw_line);
if line.is_empty() || line.starts_with(':') || line.starts_with("event:") {
continue;
}
if let Some(rest) = line.strip_prefix("data:") {
data_lines.push(rest.trim_start().to_string());
}
}
if data_lines.is_empty() {
return Ok(Vec::new());
}
let data = data_lines.join("\n");
parse_json_event(&data, state)
}
pub(crate) fn parse_json_event(
data: &str,
state: &mut StreamState,
) -> Result<Vec<ProviderEvent>, ProviderError> {
if data == "[DONE]" {
return Ok(Vec::new());
}
let event: ResponsesStreamEvent =
serde_json::from_str(data).map_err(ProviderError::Deserialize)?;
match event {
ResponsesStreamEvent::ResponseCreated { response } => {
if state.reasoning_provenance.model.is_empty() {
state.reasoning_provenance.model = response.model.clone();
}
let model = if response.model.is_empty() {
state.reasoning_provenance.model.clone()
} else {
response.model
};
Ok(vec![
ProviderEvent::ResponseCreated,
ProviderEvent::MessageStarted {
id: response.id,
model,
role: Role::Assistant,
},
])
}
ResponsesStreamEvent::ResponseOutputItemAdded { output_index, item } => {
if let Some(id) = item.reasoning_id() {
state
.reasoning_output_indices
.insert(id.to_string(), output_index);
}
if item.reasoning_encrypted_content().is_some() {
state.reasoning_encrypted_content_seen.insert(output_index);
}
let pair_function_item_ids = !state.reasoning_output_indices.is_empty();
if let Some(kind) =
item.into_provider_start(&state.reasoning_provenance, pair_function_item_ids)
{
Ok(vec![ProviderEvent::ContentBlockStarted {
index: output_index,
kind,
}])
} else {
state.ignored_output_indices.insert(output_index);
Ok(Vec::new())
}
}
ResponsesStreamEvent::ResponseOutputTextDelta {
output_index,
delta,
..
} => {
if state.ignored_output_indices.contains(&output_index) {
return Ok(Vec::new());
}
state.text_delta_seen.insert(output_index);
Ok(vec![ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::Text(delta),
}])
}
ResponsesStreamEvent::ResponseReasoningSummaryTextDelta {
output_index,
delta,
summary_index,
} => {
let mut events = vec![ProviderEvent::ReasoningSummaryDelta {
delta: delta.clone(),
summary_index,
}];
let output_index = output_index
.filter(|output_index| !state.ignored_output_indices.contains(output_index));
if let Some(output_index) = output_index {
state.reasoning_delta_seen.insert(output_index);
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ThinkingText(delta),
});
}
Ok(events)
}
ResponsesStreamEvent::ResponseReasoningTextDelta {
output_index,
delta,
content_index,
} => {
let mut events = vec![ProviderEvent::ReasoningContentDelta {
delta: delta.clone(),
content_index,
}];
let output_index = output_index
.filter(|output_index| !state.ignored_output_indices.contains(output_index));
if let Some(output_index) = output_index {
state.reasoning_delta_seen.insert(output_index);
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ThinkingText(delta),
});
}
Ok(events)
}
ResponsesStreamEvent::ResponseReasoningSummaryPartAdded { summary_index, .. } => {
Ok(vec![ProviderEvent::ReasoningSummaryPartAdded {
summary_index,
}])
}
ResponsesStreamEvent::ResponseReasoningSummaryPartDone { output_index } => {
let output_index = output_index
.filter(|output_index| !state.ignored_output_indices.contains(output_index));
if let Some(output_index) = output_index {
state.reasoning_delta_seen.insert(output_index);
return Ok(vec![ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ThinkingText("\n\n".to_string()),
}]);
}
Ok(Vec::new())
}
ResponsesStreamEvent::ResponseFunctionCallArgumentsDelta {
output_index,
delta,
..
} => {
if state.ignored_output_indices.contains(&output_index) {
return Ok(Vec::new());
}
state.function_delta_seen.insert(output_index);
Ok(vec![ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ToolUseInputJson(delta),
}])
}
ResponsesStreamEvent::ResponseToolSearchCallDelta {
output_index,
delta,
} => {
if state.ignored_output_indices.contains(&output_index) {
return Ok(Vec::new());
}
state.tool_search_delta_seen.insert(output_index);
Ok(vec![ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::HostedToolSearchQuery(delta),
}])
}
ResponsesStreamEvent::ResponseOutputItemDone { output_index, item } => {
if state.ignored_output_indices.remove(&output_index) {
return Ok(Vec::new());
}
let mut events = Vec::new();
if let Some(id) = item.reasoning_id() {
state
.reasoning_output_indices
.insert(id.to_string(), output_index);
}
let completed_text = if state.text_delta_seen.remove(&output_index) {
None
} else {
item.completed_text().filter(|text| !text.is_empty())
};
if let Some(text) = completed_text {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::Text(text),
});
}
let completed_arguments = if state.function_delta_seen.remove(&output_index) {
None
} else {
item.completed_arguments()
.filter(|arguments| !arguments.is_empty())
};
if let Some(arguments) = completed_arguments {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ToolUseInputJson(arguments),
});
}
let completed_query = if state.tool_search_delta_seen.remove(&output_index) {
None
} else {
item.completed_tool_search_query()
.filter(|query| !query.is_empty())
};
if let Some(query) = completed_query {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::HostedToolSearchQuery(query),
});
}
let reasoning = if state.reasoning_delta_seen.remove(&output_index) {
None
} else {
item.completed_reasoning_text()
.filter(|reasoning| !reasoning.is_empty())
};
if let Some(reasoning) = reasoning {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ThinkingText(reasoning),
});
}
if item.reasoning_encrypted_content().is_some() {
state.reasoning_encrypted_content_seen.insert(output_index);
}
events.extend(item.completion_deltas(output_index));
if item.is_supported() {
events.push(ProviderEvent::ContentBlockStopped {
index: output_index,
});
}
Ok(events)
}
ResponsesStreamEvent::ResponseCompleted { response }
| ResponsesStreamEvent::ResponseIncomplete { response } => {
let mut events = state.backfill_reasoning_encrypted_content(&response.output);
events.push(ProviderEvent::MessageDelta {
stop_reason: response.stop_reason(),
usage: response.usage(),
});
events.push(ProviderEvent::MessageStopped);
Ok(events)
}
ResponsesStreamEvent::ResponseFailed { response } => {
Err(ProviderError::MalformedStream(format!(
"responses response failed{}",
response
.error_message()
.map(|message| format!(": {message}"))
.unwrap_or_default()
)))
}
ResponsesStreamEvent::Error { message, error } => Err(ProviderError::MalformedStream(
error
.and_then(|error| error.message)
.or(message)
.unwrap_or_else(|| "responses stream error".to_string()),
)),
ResponsesStreamEvent::Unknown => Ok(Vec::new()),
}
}
fn find_frame_boundary(buffer: &[u8]) -> Option<(usize, usize)> {
for (index, window) in buffer.windows(2).enumerate() {
if window == b"\n\n" {
return Some((index, 2));
}
}
for (index, window) in buffer.windows(4).enumerate() {
if window == b"\r\n\r\n" {
return Some((index, 4));
}
}
None
}
#[derive(Deserialize)]
#[serde(tag = "type")]
enum ResponsesStreamEvent {
#[serde(rename = "response.created")]
ResponseCreated { response: ResponsesResponseEnvelope },
#[serde(rename = "response.output_item.added")]
ResponseOutputItemAdded {
output_index: usize,
item: ResponsesOutputItem,
},
#[serde(rename = "response.output_text.delta")]
ResponseOutputTextDelta {
output_index: usize,
delta: String,
#[allow(dead_code)]
content_index: Option<usize>,
},
#[serde(rename = "response.reasoning_summary_text.delta")]
ResponseReasoningSummaryTextDelta {
#[serde(default)]
output_index: Option<usize>,
delta: String,
summary_index: i64,
},
#[serde(rename = "response.reasoning_text.delta")]
ResponseReasoningTextDelta {
#[serde(default)]
output_index: Option<usize>,
delta: String,
content_index: i64,
},
#[serde(rename = "response.reasoning_summary_part.added")]
ResponseReasoningSummaryPartAdded {
#[serde(default, rename = "output_index")]
_output_index: Option<usize>,
summary_index: i64,
},
#[serde(rename = "response.reasoning_summary_part.done")]
ResponseReasoningSummaryPartDone {
#[serde(default)]
output_index: Option<usize>,
},
#[serde(rename = "response.function_call_arguments.delta")]
ResponseFunctionCallArgumentsDelta {
output_index: usize,
delta: String,
#[allow(dead_code)]
item_id: Option<String>,
},
#[serde(rename = "response.tool_search_call.delta")]
ResponseToolSearchCallDelta { output_index: usize, delta: String },
#[serde(rename = "response.output_item.done")]
ResponseOutputItemDone {
output_index: usize,
item: ResponsesOutputItem,
},
#[serde(rename = "response.completed")]
ResponseCompleted { response: ResponsesResponseEnvelope },
#[serde(rename = "response.incomplete")]
ResponseIncomplete { response: ResponsesResponseEnvelope },
#[serde(rename = "response.failed")]
ResponseFailed { response: ResponsesResponseEnvelope },
Error {
#[serde(default)]
message: Option<String>,
#[serde(default)]
error: Option<ResponsesErrorBody>,
},
#[serde(other)]
Unknown,
}
#[derive(Deserialize)]
struct ResponsesResponseEnvelope {
#[serde(default)]
id: String,
#[serde(default)]
model: String,
#[serde(default)]
status: Option<String>,
#[serde(default)]
usage: Option<ResponsesUsage>,
#[serde(default)]
incomplete_details: Option<ResponsesIncompleteDetails>,
#[serde(default)]
error: Option<ResponsesErrorBody>,
#[serde(default)]
output: Vec<ResponsesOutputItem>,
}
impl ResponsesResponseEnvelope {
fn stop_reason(&self) -> Option<String> {
if let Some(reason) = self
.incomplete_details
.as_ref()
.and_then(|details| details.reason.as_ref())
{
return Some(reason.clone());
}
match self.status.as_deref() {
Some("completed") | Some("in_progress") => None,
Some(status) => Some(status.to_string()),
None => None,
}
}
fn error_message(&self) -> Option<String> {
self.error.as_ref().and_then(|error| error.message.clone())
}
fn usage(&self) -> Option<TokenUsage> {
self.usage.as_ref().and_then(ResponsesUsage::to_token_usage)
}
}
#[derive(Deserialize)]
struct ResponsesIncompleteDetails {
#[serde(default)]
reason: Option<String>,
}
#[derive(Deserialize)]
struct ResponsesErrorBody {
#[serde(default)]
message: Option<String>,
}
#[derive(Deserialize)]
struct ResponsesUsage {
#[serde(default)]
input_tokens: Option<u64>,
#[serde(default)]
output_tokens: Option<u64>,
#[serde(default)]
total_tokens: Option<u64>,
#[serde(default)]
input_tokens_details: Option<ResponsesInputTokenDetails>,
#[serde(default)]
output_tokens_details: Option<ResponsesOutputTokenDetails>,
}
impl ResponsesUsage {
fn to_token_usage(&self) -> Option<TokenUsage> {
let usage = TokenUsage {
input_tokens: self.input_tokens,
output_tokens: self.output_tokens,
total_tokens: self.total_tokens,
cache_read_input_tokens: self
.input_tokens_details
.as_ref()
.and_then(|details| details.cached_tokens),
cache_creation_input_tokens: None,
reasoning_tokens: self
.output_tokens_details
.as_ref()
.and_then(|details| details.reasoning_tokens),
thoughts_tokens: None,
tool_input_tokens: None,
};
(!usage.is_empty()).then_some(usage)
}
}
#[derive(Deserialize)]
struct ResponsesInputTokenDetails {
#[serde(default)]
cached_tokens: Option<u64>,
}
#[derive(Deserialize)]
struct ResponsesOutputTokenDetails {
#[serde(default)]
reasoning_tokens: Option<u64>,
}
#[derive(Deserialize)]
#[serde(tag = "type")]
enum ResponsesOutputItem {
#[serde(rename = "message")]
Message {
#[serde(default)]
content: Vec<ResponsesMessageContent>,
},
#[serde(rename = "reasoning")]
Reasoning {
id: String,
#[serde(default)]
encrypted_content: Option<String>,
#[serde(default)]
summary: Vec<ResponsesReasoningText>,
#[serde(default)]
content: Vec<ResponsesReasoningText>,
},
#[serde(rename = "function_call")]
FunctionCall {
#[serde(default)]
id: Option<String>,
call_id: String,
name: String,
#[serde(default)]
arguments: String,
},
#[serde(rename = "tool_search_call")]
ToolSearchCall {
#[serde(default)]
id: Option<String>,
#[serde(default)]
call_id: Option<String>,
#[serde(default)]
status: Option<String>,
#[serde(default)]
_execution: Option<String>,
#[serde(default)]
arguments: Option<serde_json::Value>,
},
#[serde(rename = "web_search_call")]
WebSearchCall {
#[serde(default)]
id: Option<String>,
#[serde(default)]
status: Option<String>,
#[serde(default)]
action: Option<WebSearchAction>,
},
#[serde(rename = "image_generation_call")]
ImageGenerationCall {
id: String,
status: String,
#[serde(default)]
revised_prompt: Option<String>,
#[serde(default)]
result: Option<String>,
},
#[serde(other)]
Unsupported,
}
impl ResponsesOutputItem {
fn into_provider_start(
self,
reasoning_provenance: &ReasoningProvenance,
pair_function_item_ids: bool,
) -> Option<ContentBlockStart> {
match self {
ResponsesOutputItem::Message { .. } => Some(ContentBlockStart::Text),
ResponsesOutputItem::Reasoning {
id,
encrypted_content,
..
} => Some(ContentBlockStart::Thinking {
encrypted_content,
id: Some(id),
provenance: Some(reasoning_provenance.clone()),
redacted: false,
}),
ResponsesOutputItem::FunctionCall {
id, call_id, name, ..
} => {
let item_id = if pair_function_item_ids {
id.as_deref()
} else {
None
};
let id = encode_responses_tool_use_id(&call_id, item_id);
Some(ContentBlockStart::ToolUse { id, name })
}
ResponsesOutputItem::ToolSearchCall {
id,
call_id,
status,
arguments,
..
} => Some(ContentBlockStart::HostedToolSearch {
call: HostedToolSearchCall {
id: call_id
.or(id)
.unwrap_or_else(|| "tool_search_call".to_string()),
status,
query: arguments
.as_ref()
.and_then(extract_tool_search_query_from_value),
},
}),
ResponsesOutputItem::WebSearchCall { id, status, action } => {
Some(ContentBlockStart::HostedWebSearch {
call: HostedWebSearchCall {
id: id.unwrap_or_else(|| "web_search_call".to_string()),
status,
action,
},
})
}
ResponsesOutputItem::ImageGenerationCall {
id,
status,
revised_prompt,
result,
} => Some(ContentBlockStart::ImageGeneration {
call: ImageGenerationCall {
id,
status,
revised_prompt,
result: result.map(|result| ImageGenerationResult::ArtifactRef {
artifact_id: result,
}),
},
}),
ResponsesOutputItem::Unsupported => None,
}
}
fn completed_text(&self) -> Option<String> {
match self {
ResponsesOutputItem::Message { content } => {
let text = content
.iter()
.filter_map(ResponsesMessageContent::text)
.collect::<Vec<_>>()
.join("");
Some(text)
}
_ => None,
}
}
fn completed_reasoning_text(&self) -> Option<String> {
match self {
ResponsesOutputItem::Reasoning {
summary, content, ..
} => {
let summary = summary
.iter()
.map(|part| part.text.as_str())
.collect::<Vec<_>>()
.join("\n\n");
if !summary.is_empty() {
return Some(summary);
}
Some(
content
.iter()
.map(|part| part.text.as_str())
.collect::<Vec<_>>()
.join("\n\n"),
)
}
_ => None,
}
}
fn reasoning_id(&self) -> Option<&str> {
match self {
ResponsesOutputItem::Reasoning { id, .. } => Some(id),
_ => None,
}
}
fn reasoning_encrypted_content(&self) -> Option<(&str, &str)> {
match self {
ResponsesOutputItem::Reasoning {
id,
encrypted_content: Some(encrypted_content),
..
} if !encrypted_content.is_empty() => Some((id, encrypted_content)),
_ => None,
}
}
fn completed_arguments(&self) -> Option<String> {
match self {
ResponsesOutputItem::FunctionCall { arguments, .. } => Some(arguments.clone()),
_ => None,
}
}
fn completed_tool_search_query(&self) -> Option<String> {
match self {
ResponsesOutputItem::ToolSearchCall { arguments, .. } => arguments
.as_ref()
.and_then(extract_tool_search_query_from_value),
_ => None,
}
}
fn completion_deltas(&self, output_index: usize) -> Vec<ProviderEvent> {
match self {
ResponsesOutputItem::ToolSearchCall { status, .. } => status
.clone()
.filter(|status| !status.is_empty())
.map(|status| ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::HostedToolSearchStatus(status),
})
.into_iter()
.collect(),
ResponsesOutputItem::Reasoning {
encrypted_content: Some(encrypted_content),
..
} if !encrypted_content.is_empty() => {
vec![ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ThinkingEncryptedContent(encrypted_content.clone()),
}]
}
ResponsesOutputItem::WebSearchCall { status, action, .. } => {
let mut events = Vec::new();
if let Some(action) = action.clone() {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::HostedWebSearchAction(action),
});
}
if let Some(status) = status.clone().filter(|status| !status.is_empty()) {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::HostedWebSearchStatus(status),
});
}
events
}
ResponsesOutputItem::ImageGenerationCall {
status,
revised_prompt,
result,
..
} => {
let mut events = Vec::new();
if let Some(revised_prompt) = revised_prompt
.clone()
.filter(|revised_prompt| !revised_prompt.is_empty())
{
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ImageGenerationRevisedPrompt(revised_prompt),
});
}
if let Some(result) = result.clone() {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ImageGenerationResult(
ImageGenerationResult::ArtifactRef {
artifact_id: result,
},
),
});
}
if !status.is_empty() {
events.push(ProviderEvent::ContentBlockDelta {
index: output_index,
delta: ContentBlockDelta::ImageGenerationStatus(status.clone()),
});
}
events
}
_ => Vec::new(),
}
}
fn is_supported(&self) -> bool {
!matches!(self, ResponsesOutputItem::Unsupported)
}
}
#[derive(Deserialize)]
struct ResponsesReasoningText {
#[serde(default)]
text: String,
}
fn extract_tool_search_query_from_value(value: &serde_json::Value) -> Option<String> {
value
.get("query")
.and_then(serde_json::Value::as_str)
.map(str::to_string)
}
#[derive(Deserialize)]
#[serde(tag = "type")]
enum ResponsesMessageContent {
#[serde(rename = "output_text")]
OutputText { text: String },
#[serde(rename = "input_text")]
InputText { text: String },
#[serde(other)]
Unsupported,
}
impl ResponsesMessageContent {
fn text(&self) -> Option<String> {
match self {
ResponsesMessageContent::OutputText { text }
| ResponsesMessageContent::InputText { text } => Some(text.clone()),
ResponsesMessageContent::Unsupported => None,
}
}
}
#[cfg(test)]
mod tests {
use http::HeaderMap;
use http::HeaderValue;
use crate::{
ContentBlock, ContentBlockDelta, ContentBlockStart, ProviderError, ProviderEvent,
ProviderId, ReasoningFormat, ReasoningProvenance, Role, TokenUsage,
};
use super::{StreamState, parse_frame, response_headers_event};
#[test]
fn emits_response_headers_event_for_metadata_headers() {
let mut headers = HeaderMap::new();
headers.insert("openai-model", HeaderValue::from_static("gpt-5"));
headers.insert("x-models-etag", HeaderValue::from_static("etag-123"));
headers.insert(
"x-ratelimit-limit-requests",
HeaderValue::from_static("1000"),
);
let event = response_headers_event(&headers).expect("headers event should be emitted");
assert_eq!(
event,
ProviderEvent::ResponseHeaders(crate::ResponseHeaders {
values: vec![
("openai-model".to_string(), "gpt-5".to_string()),
("x-models-etag".to_string(), "etag-123".to_string()),
("x-ratelimit-limit-requests".to_string(), "1000".to_string()),
],
})
);
}
#[test]
fn parses_reasoning_delta_stream_events() {
let mut state = StreamState::default();
let summary_delta = parse_frame(
br#"data: {"type":"response.reasoning_summary_text.delta","summary_index":2,"delta":"short summary"}"#,
&mut state,
)
.expect("reasoning summary delta should parse");
assert_eq!(
summary_delta,
vec![ProviderEvent::ReasoningSummaryDelta {
delta: "short summary".to_string(),
summary_index: 2,
}]
);
let part_added = parse_frame(
br#"data: {"type":"response.reasoning_summary_part.added","summary_index":2}"#,
&mut state,
)
.expect("reasoning summary part should parse");
assert_eq!(
part_added,
vec![ProviderEvent::ReasoningSummaryPartAdded { summary_index: 2 }]
);
let reasoning_delta = parse_frame(
br#"data: {"type":"response.reasoning_text.delta","content_index":7,"delta":"internal chain"}"#,
&mut state,
)
.expect("reasoning content delta should parse");
assert_eq!(
reasoning_delta,
vec![ProviderEvent::ReasoningContentDelta {
delta: "internal chain".to_string(),
content_index: 7,
}]
);
}
#[test]
fn captures_reasoning_items_and_pairs_function_item_ids() {
let mut state =
StreamState::new(ProviderId::new("openai-edge"), "gpt-requested".to_string());
let added = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_1","encrypted_content":null,"summary":[]}}"#,
&mut state,
)
.expect("reasoning item start should parse");
assert_eq!(
added,
vec![ProviderEvent::ContentBlockStarted {
index: 0,
kind: ContentBlockStart::Thinking {
encrypted_content: None,
id: Some("rs_1".to_string()),
provenance: Some(ReasoningProvenance {
provider: ProviderId::new("openai-edge"),
model: "gpt-requested".to_string(),
format: ReasoningFormat::OpenAiEncrypted,
}),
redacted: false,
},
}]
);
let delta = parse_frame(
br#"data: {"type":"response.reasoning_summary_text.delta","output_index":0,"summary_index":0,"delta":"short summary"}"#,
&mut state,
)
.expect("reasoning summary delta should parse");
assert_eq!(
delta,
vec![
ProviderEvent::ReasoningSummaryDelta {
delta: "short summary".to_string(),
summary_index: 0,
},
ProviderEvent::ContentBlockDelta {
index: 0,
delta: ContentBlockDelta::ThinkingText("short summary".to_string()),
},
]
);
let done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rs_1","encrypted_content":"encrypted-1","summary":[{"type":"summary_text","text":"short summary"}]}}"#,
&mut state,
)
.expect("reasoning item completion should parse");
assert_eq!(
done,
vec![
ProviderEvent::ContentBlockDelta {
index: 0,
delta: ContentBlockDelta::ThinkingEncryptedContent("encrypted-1".to_string()),
},
ProviderEvent::ContentBlockStopped { index: 0 },
]
);
let tool = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"read_file","arguments":""}}"#,
&mut state,
)
.expect("function call should parse");
assert_eq!(
tool,
vec![ProviderEvent::ContentBlockStarted {
index: 1,
kind: ContentBlockStart::ToolUse {
id: "call_1|fc_1".to_string(),
name: "read_file".to_string(),
},
}]
);
}
#[test]
fn function_item_id_stays_out_of_non_reasoning_tool_use_ids() {
let mut state = StreamState::default();
let tool = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","id":"fc_1","call_id":"call_1","name":"read_file","arguments":""}}"#,
&mut state,
)
.expect("function call should parse");
assert_eq!(
tool,
vec![ProviderEvent::ContentBlockStarted {
index: 0,
kind: ContentBlockStart::ToolUse {
id: "call_1".to_string(),
name: "read_file".to_string(),
},
}]
);
}
#[tokio::test]
async fn completed_response_backfills_late_azure_encrypted_content() {
let mut state =
StreamState::new(ProviderId::new("azure-openai"), "gpt-requested".to_string());
let frames: &[&[u8]] = &[
br#"data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-resolved","status":"in_progress"}}"#,
br#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_1","summary":[]}}"#,
br#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rs_1","summary":[{"type":"summary_text","text":"summary"}]}}"#,
br#"data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-resolved","status":"completed","output":[{"type":"reasoning","id":"rs_1","encrypted_content":"from-completed","summary":[{"type":"summary_text","text":"summary"}]}]}}"#,
];
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
for frame in frames {
for event in parse_frame(frame, &mut state).expect("frame should parse") {
tx.send(Ok(event)).unwrap();
}
}
drop(tx);
let response = crate::collect_response_from_stream(rx)
.await
.expect("response should rebuild after metadata backfill");
assert_eq!(
response.content,
vec![ContentBlock::Thinking {
thinking: "summary".to_string(),
signature: None,
encrypted_content: Some("from-completed".to_string()),
id: Some("rs_1".to_string()),
provenance: Some(ReasoningProvenance {
provider: ProviderId::new("azure-openai"),
model: "gpt-requested".to_string(),
format: ReasoningFormat::OpenAiEncrypted,
}),
redacted: false,
}]
);
}
#[tokio::test]
async fn completed_response_does_not_replace_output_item_encrypted_content() {
let mut state =
StreamState::new(ProviderId::new("azure-openai"), "gpt-requested".to_string());
let frames: &[&[u8]] = &[
br#"data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-resolved","status":"in_progress"}}"#,
br#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"reasoning","id":"rs_1","summary":[]}}"#,
br#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"reasoning","id":"rs_1","encrypted_content":"from-done","summary":[]}}"#,
br#"data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-resolved","status":"completed","output":[{"type":"reasoning","id":"rs_1","encrypted_content":"from-completed","summary":[]}]}}"#,
];
let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
for frame in frames {
for event in parse_frame(frame, &mut state).expect("frame should parse") {
tx.send(Ok(event)).unwrap();
}
}
drop(tx);
let response = crate::collect_response_from_stream(rx)
.await
.expect("response should rebuild");
let ContentBlock::Thinking {
encrypted_content, ..
} = &response.content[0]
else {
panic!("expected thinking block");
};
assert_eq!(encrypted_content.as_deref(), Some("from-done"));
}
#[test]
fn parses_tool_call_stream_events() {
let mut state = StreamState::default();
let created = parse_frame(
br#"data: {"type":"response.created","response":{"id":"resp_1","model":"gpt-5","status":"in_progress"}}"#,
&mut state,
)
.expect("created event should parse");
assert_eq!(
created,
vec![
ProviderEvent::ResponseCreated,
ProviderEvent::MessageStarted {
id: "resp_1".to_string(),
model: "gpt-5".to_string(),
role: Role::Assistant,
},
]
);
let added = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"read_file","arguments":""}}"#,
&mut state,
)
.expect("tool call start should parse");
assert_eq!(
added,
vec![ProviderEvent::ContentBlockStarted {
index: 0,
kind: ContentBlockStart::ToolUse {
id: "call_1".to_string(),
name: "read_file".to_string(),
},
}]
);
let delta = parse_frame(
br#"data: {"type":"response.function_call_arguments.delta","output_index":0,"delta":"{\"path\":\"README.md\"}"}"#,
&mut state,
)
.expect("tool arguments delta should parse");
assert_eq!(
delta,
vec![ProviderEvent::ContentBlockDelta {
index: 0,
delta: ContentBlockDelta::ToolUseInputJson("{\"path\":\"README.md\"}".to_string()),
}]
);
let done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":0,"item":{"type":"function_call","call_id":"call_1","name":"read_file","arguments":"{\"path\":\"README.md\"}"}}"#,
&mut state,
)
.expect("tool call completion should parse");
assert_eq!(done, vec![ProviderEvent::ContentBlockStopped { index: 0 }]);
}
#[test]
fn falls_back_to_completed_message_text_when_no_text_delta_arrives() {
let mut state = StreamState::default();
let _ = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":1,"item":{"type":"message","content":[]}}"#,
&mut state,
)
.expect("message start should parse");
let done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":1,"item":{"type":"message","content":[{"type":"output_text","text":"Hello"}]}}"#,
&mut state,
)
.expect("message completion should parse");
assert_eq!(
done,
vec![
ProviderEvent::ContentBlockDelta {
index: 1,
delta: ContentBlockDelta::Text("Hello".to_string()),
},
ProviderEvent::ContentBlockStopped { index: 1 },
]
);
let completed = parse_frame(
br#"data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-5","status":"completed"}}"#,
&mut state,
)
.expect("completion should parse");
assert_eq!(
completed,
vec![
ProviderEvent::MessageDelta {
stop_reason: None,
usage: None,
},
ProviderEvent::MessageStopped,
]
);
}
#[test]
fn parses_final_usage_from_completed_response() {
let mut state = StreamState::default();
let completed = parse_frame(
br#"data: {"type":"response.completed","response":{"id":"resp_1","model":"gpt-5","status":"completed","usage":{"input_tokens":328,"input_tokens_details":{"cached_tokens":12},"output_tokens":52,"output_tokens_details":{"reasoning_tokens":7},"total_tokens":380}}}"#,
&mut state,
)
.expect("completion should parse");
assert_eq!(
completed,
vec![
ProviderEvent::MessageDelta {
stop_reason: None,
usage: Some(TokenUsage {
input_tokens: Some(328),
output_tokens: Some(52),
total_tokens: Some(380),
cache_read_input_tokens: Some(12),
cache_creation_input_tokens: None,
reasoning_tokens: Some(7),
thoughts_tokens: None,
tool_input_tokens: None,
}),
},
ProviderEvent::MessageStopped,
]
);
}
#[test]
fn surfaces_the_provider_error_from_a_failed_response_without_a_model() {
let mut state = StreamState::default();
let error = parse_frame(
br#"data: {"type":"response.failed","response":{"error":{"message":"upstream exploded"}}}"#,
&mut state,
)
.expect_err("failed response should surface the provider error");
assert!(
matches!(&error, ProviderError::MalformedStream(message) if message.contains("upstream exploded")),
"unexpected error: {error}"
);
}
#[test]
fn terminates_the_message_for_an_incomplete_response_without_a_model() {
let mut state = StreamState::default();
let incomplete = parse_frame(
br#"data: {"type":"response.incomplete","response":{"status":"max_output_tokens"}}"#,
&mut state,
)
.expect("incomplete response should parse");
assert_eq!(
incomplete,
vec![
ProviderEvent::MessageDelta {
stop_reason: Some("max_output_tokens".to_string()),
usage: None,
},
ProviderEvent::MessageStopped,
]
);
}
#[test]
fn falls_back_to_the_requested_model_when_a_created_response_omits_it() {
let mut state = StreamState::new(ProviderId::new("openai"), "gpt-requested".to_string());
let created = parse_frame(
br#"data: {"type":"response.created","response":{"id":"resp_1"}}"#,
&mut state,
)
.expect("created event should parse");
assert_eq!(
created,
vec![
ProviderEvent::ResponseCreated,
ProviderEvent::MessageStarted {
id: "resp_1".to_string(),
model: "gpt-requested".to_string(),
role: Role::Assistant,
},
]
);
assert_eq!(state.reasoning_provenance.model, "gpt-requested");
}
#[test]
fn parses_hosted_tool_search_output_items() {
let mut state = StreamState::default();
let added = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":3,"item":{"type":"tool_search_call","id":"search_1","status":"in_progress"}}"#,
&mut state,
)
.expect("hosted search start should parse");
assert_eq!(
added,
vec![ProviderEvent::ContentBlockStarted {
index: 3,
kind: ContentBlockStart::HostedToolSearch {
call: crate::HostedToolSearchCall {
id: "search_1".to_string(),
status: Some("in_progress".to_string()),
query: None,
},
},
}]
);
let delta = parse_frame(
br#"data: {"type":"response.tool_search_call.delta","output_index":3,"delta":"weather"}"#,
&mut state,
)
.expect("hosted search delta should parse");
assert_eq!(
delta,
vec![ProviderEvent::ContentBlockDelta {
index: 3,
delta: ContentBlockDelta::HostedToolSearchQuery("weather".to_string()),
}]
);
let done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":3,"item":{"type":"tool_search_call","id":"search_1","status":"completed","arguments":{"query":"weather"}}}"#,
&mut state,
)
.expect("hosted search completion should parse");
assert_eq!(
done,
vec![
ProviderEvent::ContentBlockDelta {
index: 3,
delta: ContentBlockDelta::HostedToolSearchStatus("completed".to_string()),
},
ProviderEvent::ContentBlockStopped { index: 3 },
]
);
}
#[test]
fn parses_web_search_and_image_generation_output_items() {
let mut state = StreamState::default();
let added = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":4,"item":{"type":"web_search_call","id":"ws_1","status":"in_progress"}}"#,
&mut state,
)
.expect("web search start should parse");
assert_eq!(
added,
vec![ProviderEvent::ContentBlockStarted {
index: 4,
kind: ContentBlockStart::HostedWebSearch {
call: crate::HostedWebSearchCall {
id: "ws_1".to_string(),
status: Some("in_progress".to_string()),
action: None,
},
},
}]
);
let done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":4,"item":{"type":"web_search_call","id":"ws_1","status":"completed","action":{"type":"search","query":"weather seattle"}}}"#,
&mut state,
)
.expect("web search done should parse");
assert_eq!(
done,
vec![
ProviderEvent::ContentBlockDelta {
index: 4,
delta: ContentBlockDelta::HostedWebSearchAction(
crate::WebSearchAction::Search {
query: Some("weather seattle".to_string()),
queries: None,
}
),
},
ProviderEvent::ContentBlockDelta {
index: 4,
delta: ContentBlockDelta::HostedWebSearchStatus("completed".to_string()),
},
ProviderEvent::ContentBlockStopped { index: 4 },
]
);
let image_added = parse_frame(
br#"data: {"type":"response.output_item.added","output_index":5,"item":{"type":"image_generation_call","id":"ig_1","status":"in_progress"}}"#,
&mut state,
)
.expect("image generation start should parse");
assert_eq!(
image_added,
vec![ProviderEvent::ContentBlockStarted {
index: 5,
kind: ContentBlockStart::ImageGeneration {
call: crate::ImageGenerationCall {
id: "ig_1".to_string(),
status: "in_progress".to_string(),
revised_prompt: None,
result: None,
},
},
}]
);
let image_done = parse_frame(
br#"data: {"type":"response.output_item.done","output_index":5,"item":{"type":"image_generation_call","id":"ig_1","status":"completed","revised_prompt":"A blue square","result":"artifact_1"}}"#,
&mut state,
)
.expect("image generation done should parse");
assert_eq!(
image_done,
vec![
ProviderEvent::ContentBlockDelta {
index: 5,
delta: ContentBlockDelta::ImageGenerationRevisedPrompt(
"A blue square".to_string()
),
},
ProviderEvent::ContentBlockDelta {
index: 5,
delta: ContentBlockDelta::ImageGenerationResult(
crate::ImageGenerationResult::ArtifactRef {
artifact_id: "artifact_1".to_string(),
}
),
},
ProviderEvent::ContentBlockDelta {
index: 5,
delta: ContentBlockDelta::ImageGenerationStatus("completed".to_string()),
},
ProviderEvent::ContentBlockStopped { index: 5 },
]
);
}
}