use crate::error_display;
use crate::provider;
use crate::providers::shared::StreamTelemetry;
use crate::providers::shared::parse_cache_write_tokens_from_usage;
use crate::providers::shared::parse_cached_prompt_tokens_from_usage;
use crate::providers::shared::{
ResponsesStreamEventPolicy, StreamAssemblyError, extract_data_payload, response_stream_event_policy,
};
use async_stream::try_stream;
use futures::StreamExt;
use hashbrown::{HashMap, HashSet};
use serde_json::{Value, json};
use std::time::Instant;
use vtcode_commons::model_family::find_family_for_model;
use super::responses_api::parse_responses_payload;
use super::streaming::OpenAIStreamTelemetry;
fn strip_reasoning_for_model(model: &str, mut response: provider::LLMResponse) -> provider::LLMResponse {
if !find_family_for_model(model).supports_reasoning_summaries {
response.reasoning = None;
response.reasoning_details = None;
}
response
}
fn streamed_response_is_usable(response: &provider::LLMResponse) -> bool {
response.content.as_deref().is_some_and(|content| !content.is_empty())
|| response.tool_calls.as_ref().is_some_and(|tool_calls| !tool_calls.is_empty())
|| response.reasoning.as_deref().is_some_and(|reasoning| !reasoning.is_empty())
|| response.reasoning_details.as_ref().is_some_and(|details| !details.is_empty())
}
fn final_response_output_is_empty(final_response: &Value) -> bool {
final_response
.get("output")
.and_then(Value::as_array)
.is_some_and(Vec::is_empty)
}
fn merge_final_response_metadata(
response: &mut provider::LLMResponse,
final_response: &Value,
include_cached_prompt_metrics: bool,
) {
if let Some(usage_value) = final_response.get("usage") {
let cached_prompt_tokens = parse_cached_prompt_tokens_from_usage(usage_value, include_cached_prompt_metrics);
let cache_creation_tokens = parse_cache_write_tokens_from_usage(usage_value, include_cached_prompt_metrics);
response.usage = Some(provider::Usage {
prompt_tokens: usage_value
.get("input_tokens")
.or_else(|| usage_value.get("prompt_tokens"))
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(0),
completion_tokens: usage_value
.get("output_tokens")
.or_else(|| usage_value.get("completion_tokens"))
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(0),
total_tokens: usage_value
.get("total_tokens")
.and_then(Value::as_u64)
.and_then(|value| u32::try_from(value).ok())
.unwrap_or(0),
cached_prompt_tokens,
cache_creation_tokens,
cache_read_tokens: None,
iterations: None,
});
}
if let Some(request_id) = final_response
.get("id")
.and_then(Value::as_str)
.or_else(|| final_response.get("request_id").and_then(Value::as_str))
{
response.request_id = Some(request_id.to_string());
}
}
#[derive(Default)]
struct ResponsesToolCallState {
item_id_to_call_id: HashMap<String, String>,
tool_call_indexes: HashMap<String, usize>,
next_tool_call_index: usize,
fabricated_ids: HashMap<usize, String>,
fabricated_fallback_id: Option<String>,
}
impl ResponsesToolCallState {
fn capture_metadata(
&mut self,
aggregator: &mut crate::providers::shared::StreamAggregator,
item: &Value,
output_index: Option<usize>,
) {
if item.get("type").and_then(Value::as_str) != Some("function_call") {
return;
}
let item_id = item.get("id").and_then(Value::as_str).filter(|value| !value.is_empty());
let provider_call_id = item.get("call_id").and_then(Value::as_str).filter(|value| !value.is_empty());
let Some(call_id) = provider_call_id.or(item_id) else {
return;
};
let Some(name) = item.get("name").and_then(Value::as_str).or_else(|| {
item.get("function")
.and_then(|function| function.get("name"))
.and_then(Value::as_str)
}) else {
return;
};
self.capture_item_call_id_mapping(item_id, provider_call_id);
let output_index =
output_index.or_else(|| item_id.and_then(|item_id| self.tool_call_indexes.get(item_id).copied()));
let index = self.resolve_tool_call_index(call_id, output_index);
aggregator.handle_tool_calls(&[json!({
"index": index,
"id": call_id,
"function": {
"name": name,
}
})]);
}
fn handle_arguments_delta(
&mut self,
aggregator: &mut crate::providers::shared::StreamAggregator,
payload: &Value,
) -> Result<(), provider::LLMError> {
let delta = payload
.get("delta")
.and_then(Value::as_str)
.ok_or_else(|| StreamAssemblyError::MissingField("delta").into_llm_error("OpenAI"))?;
let item_id = payload.get("item_id").and_then(Value::as_str);
let payload_call_id = payload.get("call_id").and_then(Value::as_str);
self.capture_item_call_id_mapping(item_id, payload_call_id);
let output_index = payload.get("output_index").and_then(Value::as_u64).map(|value| value as usize);
let call_id = match self.resolve_provider_call_id(item_id, payload_call_id) {
Some(call_id) => call_id,
None => self.fabricated_call_id(output_index),
};
let index = self.resolve_tool_call_index(&call_id, output_index);
if !delta.is_empty() {
aggregator.handle_tool_calls(&[json!({
"index": index,
"id": call_id,
"function": {
"arguments": delta,
}
})]);
}
Ok(())
}
fn capture_item_call_id_mapping(&mut self, item_id: Option<&str>, call_id: Option<&str>) {
let Some(item_id) = item_id.filter(|value| !value.is_empty()) else {
return;
};
let Some(call_id) = call_id.filter(|value| !value.is_empty()) else {
return;
};
self.item_id_to_call_id.insert(item_id.to_string(), call_id.to_string());
}
fn resolve_provider_call_id(&self, item_id: Option<&str>, call_id: Option<&str>) -> Option<String> {
call_id
.filter(|value| !value.is_empty())
.or_else(|| item_id.and_then(|value| self.item_id_to_call_id.get(value).map(String::as_str)))
.or_else(|| item_id.filter(|value| !value.is_empty()))
.map(ToOwned::to_owned)
}
fn fabricated_call_id(&mut self, output_index: Option<usize>) -> String {
match output_index {
Some(index) => self
.fabricated_ids
.entry(index)
.or_insert_with(crate::providers::shared::generate_tool_call_id)
.clone(),
None => self
.fabricated_fallback_id
.get_or_insert_with(crate::providers::shared::generate_tool_call_id)
.clone(),
}
}
fn resolve_tool_call_index(&mut self, call_id: &str, output_index: Option<usize>) -> usize {
if let Some(index) = output_index {
self.tool_call_indexes.insert(call_id.to_string(), index);
self.next_tool_call_index = self.next_tool_call_index.max(index + 1);
return index;
}
if let Some(index) = self.tool_call_indexes.get(call_id).copied() {
return index;
}
let index = self.next_tool_call_index;
self.tool_call_indexes.insert(call_id.to_string(), index);
self.next_tool_call_index += 1;
index
}
}
enum ChatStreamEvent {
TextDelta {
delta: String,
},
ReasoningDelta {
delta: String,
source: provider::ReasoningSource,
},
ReasoningStage {
stage: String,
},
ToolCallStart {
call_id: String,
name: Option<String>,
},
ToolCallDelta {
call_id: String,
delta: String,
},
Completed {
response: Box<provider::LLMResponse>,
},
}
#[derive(Default)]
struct ChatToolCallState {
call_ids_by_index: HashMap<usize, String>,
started_call_ids: HashSet<String>,
}
impl ChatToolCallState {
fn handle_deltas(
&mut self,
aggregator: &mut crate::providers::shared::StreamAggregator,
tool_deltas: &[Value],
) -> Vec<ChatStreamEvent> {
let mut events = Vec::new();
for (position, tool_delta) in tool_deltas.iter().enumerate() {
let Some(tool_delta_object) = tool_delta.as_object() else {
continue;
};
let index = tool_delta_object
.get("index")
.and_then(Value::as_u64)
.map(|value| value as usize)
.unwrap_or(position);
let provider_call_id = tool_delta_object
.get("id")
.and_then(Value::as_str)
.filter(|value| !value.is_empty());
let call_id = if let Some(call_id) = self.call_ids_by_index.get(&index) {
call_id.clone()
} else {
let call_id = provider_call_id
.map(ToOwned::to_owned)
.unwrap_or_else(crate::providers::shared::generate_tool_call_id);
self.call_ids_by_index.insert(index, call_id.clone());
call_id
};
let mut patched_tool_delta = tool_delta.clone();
if let Some(object) = patched_tool_delta.as_object_mut() {
object.insert("index".to_string(), Value::from(index as u64));
object.insert("id".to_string(), Value::String(call_id.clone()));
}
aggregator.handle_tool_calls(std::slice::from_ref(&patched_tool_delta));
let name = tool_delta_object
.get("function")
.and_then(Value::as_object)
.and_then(|function| function.get("name"))
.and_then(Value::as_str)
.filter(|value| !value.is_empty())
.map(ToOwned::to_owned);
if self.started_call_ids.insert(call_id.clone()) {
events.push(ChatStreamEvent::ToolCallStart { call_id: call_id.clone(), name });
}
if let Some(arguments) = tool_delta_object
.get("function")
.and_then(Value::as_object)
.and_then(|function| function.get("arguments"))
.and_then(tool_argument_delta)
{
events.push(ChatStreamEvent::ToolCallDelta { call_id, delta: arguments });
}
}
events
}
}
fn tool_argument_delta(value: &Value) -> Option<String> {
match value {
Value::String(value) if !value.is_empty() => Some(value.clone()),
Value::Array(_) | Value::Object(_) => Some(value.to_string()),
_ => None,
}
}
fn create_chat_event_stream(
response: reqwest::Response,
model: String,
) -> impl futures::Stream<Item = Result<ChatStreamEvent, provider::LLMError>> + Send {
let stream = try_stream! {
let mut body_stream = response.bytes_stream();
let mut buf: Vec<u8> = Vec::new();
let mut offset = 0usize;
let mut aggregator = crate::providers::shared::StreamAggregator::new(model.clone());
let mut tool_call_state = ChatToolCallState::default();
let telemetry = OpenAIStreamTelemetry;
while let Some(chunk_result) = body_stream.next().await {
let chunk = chunk_result.map_err(|err| {
let formatted_error = error_display::format_llm_error(
"OpenAI",
&format!("Streaming error: {err}"),
);
provider::LLMError::Network { message: formatted_error, metadata: None }
})?;
buf.extend_from_slice(&chunk);
while let Some((split_idx, delimiter_len)) = crate::providers::shared::find_sse_boundary_bytes(&buf, offset) {
let event = std::str::from_utf8(&buf[offset..split_idx]).expect("valid utf-8 stream data");
offset = split_idx + delimiter_len;
if let Some(data_payload) = extract_data_payload(event) {
let trimmed_payload = data_payload.trim();
if trimmed_payload.is_empty() || trimmed_payload == "[DONE]" {
continue;
}
let mut payload: Value = serde_json::from_str(trimmed_payload).map_err(|err| {
StreamAssemblyError::InvalidPayload(err.to_string())
.into_llm_error("OpenAI")
})?;
if let Some(usage_val) = payload.get_mut("usage")
&& let Ok(u) = serde_json::from_value::<provider::Usage>(std::mem::take(usage_val)) {
aggregator.set_usage(u);
}
if let Some(choices) = payload.get("choices").and_then(|v| v.as_array())
&& let Some(choice) = choices.first() {
if let Some(delta) = choice.get("delta") {
if let Some(content) = delta.get("content").and_then(|v| v.as_str()) {
telemetry.on_content_delta(content);
for event in aggregator.handle_content(content) {
match event {
provider::LLMStreamEvent::Token { delta } => {
yield ChatStreamEvent::TextDelta { delta };
}
provider::LLMStreamEvent::Reasoning { delta } => {
yield ChatStreamEvent::ReasoningDelta {
delta,
source: provider::ReasoningSource::Unknown,
};
}
provider::LLMStreamEvent::ReasoningStage { stage } => {
yield ChatStreamEvent::ReasoningStage { stage };
}
provider::LLMStreamEvent::ReasoningSignature { .. }
| provider::LLMStreamEvent::Completed { .. } => {}
}
}
}
if let Some(reasoning) = delta.get("reasoning_content").and_then(|v| v.as_str())
&& let Some(delta) = aggregator.handle_reasoning(reasoning) {
telemetry.on_reasoning_delta(&delta);
yield ChatStreamEvent::ReasoningDelta {
delta,
source: provider::ReasoningSource::Unknown,
};
}
if let Some(tool_deltas) = delta.get("tool_calls").and_then(|v| v.as_array()) {
for event in tool_call_state.handle_deltas(&mut aggregator, tool_deltas) {
yield event;
}
telemetry.on_tool_call_delta();
}
}
if let Some(reason) = choice.get("finish_reason").and_then(|v| v.as_str()) {
aggregator.set_finish_reason(match reason {
"stop" => provider::FinishReason::Stop,
"length" => provider::FinishReason::Length,
"tool_calls" => provider::FinishReason::ToolCalls,
"content_filter" => provider::FinishReason::ContentFilter,
_ => provider::FinishReason::Stop,
});
}
}
}
}
if offset > 0 {
buf.drain(..offset);
offset = 0;
}
}
let response = aggregator.finalize();
let response = strip_reasoning_for_model(&model, response);
yield ChatStreamEvent::Completed { response: Box::new(response) };
};
stream
}
fn chat_event_to_legacy(event: ChatStreamEvent) -> Option<provider::LLMStreamEvent> {
match event {
ChatStreamEvent::TextDelta { delta } => Some(provider::LLMStreamEvent::Token { delta }),
ChatStreamEvent::ReasoningDelta { delta, .. } => Some(provider::LLMStreamEvent::Reasoning { delta }),
ChatStreamEvent::ReasoningStage { stage } => Some(provider::LLMStreamEvent::ReasoningStage { stage }),
ChatStreamEvent::ToolCallStart { .. } | ChatStreamEvent::ToolCallDelta { .. } => None,
ChatStreamEvent::Completed { response } => Some(provider::LLMStreamEvent::Completed { response }),
}
}
fn chat_event_to_normalized(event: ChatStreamEvent) -> Vec<provider::NormalizedStreamEvent> {
match event {
ChatStreamEvent::TextDelta { delta } => vec![provider::NormalizedStreamEvent::TextDelta { delta }],
ChatStreamEvent::ReasoningDelta { delta, source } => {
vec![provider::NormalizedStreamEvent::ReasoningDelta { delta, source }]
}
ChatStreamEvent::ReasoningStage { stage } => vec![provider::NormalizedStreamEvent::ReasoningStage { stage }],
ChatStreamEvent::ToolCallStart { call_id, name } => {
vec![provider::NormalizedStreamEvent::ToolCallStart { call_id, name }]
}
ChatStreamEvent::ToolCallDelta { call_id, delta } => {
vec![provider::NormalizedStreamEvent::ToolCallDelta { call_id, delta }]
}
ChatStreamEvent::Completed { response } => {
let mut events = Vec::new();
if let Some(usage) = response.usage.clone() {
events.push(provider::NormalizedStreamEvent::Usage { usage });
}
events.push(provider::NormalizedStreamEvent::Done { response });
events
}
}
}
pub(crate) fn create_chat_stream(response: reqwest::Response, model: String) -> provider::LLMStream {
let mut decoded = Box::pin(create_chat_event_stream(response, model));
let stream = try_stream! {
while let Some(event) = decoded.next().await {
if let Some(event) = chat_event_to_legacy(event?) {
yield event;
}
}
};
Box::pin(stream)
}
pub(crate) fn create_chat_normalized_stream(
response: reqwest::Response,
model: String,
) -> provider::LLMNormalizedStream {
let mut decoded = Box::pin(create_chat_event_stream(response, model));
let stream = try_stream! {
while let Some(event) = decoded.next().await {
for normalized in chat_event_to_normalized(event?) {
yield normalized;
}
}
};
Box::pin(stream)
}
pub(crate) fn create_responses_stream(
response: reqwest::Response,
model: String,
include_metrics: bool,
_debug_model: Option<String>,
_request_timer: Option<Instant>,
) -> provider::LLMStream {
let stream = try_stream! {
let mut body_stream = response.bytes_stream();
let mut buf: Vec<u8> = Vec::new();
let mut offset = 0usize;
let mut aggregator = crate::providers::shared::StreamAggregator::new(model.clone());
let retain_reasoning_summaries = find_family_for_model(&model).supports_reasoning_summaries;
let mut final_response: Option<Value> = None;
let mut done = false;
let mut tool_call_state = ResponsesToolCallState::default();
#[cfg(debug_assertions)]
let mut streamed_events_counter: usize = 0;
let telemetry = OpenAIStreamTelemetry;
while let Some(chunk_result) = body_stream.next().await {
let chunk = chunk_result.map_err(|err| {
let formatted_error = error_display::format_llm_error(
"OpenAI",
&format!("Streaming error: {err}"),
);
provider::LLMError::Network { message: formatted_error, metadata: None }
})?;
buf.extend_from_slice(&chunk);
while let Some((split_idx, delimiter_len)) = crate::providers::shared::find_sse_boundary_bytes(&buf, offset) {
let event = std::str::from_utf8(&buf[offset..split_idx]).expect("valid utf-8 stream data");
offset = split_idx + delimiter_len;
#[cfg(debug_assertions)]
{
streamed_events_counter = streamed_events_counter.saturating_add(1);
}
if let Some(data_payload) = extract_data_payload(event) {
let trimmed_payload = data_payload.trim();
if trimmed_payload.is_empty() {
continue;
}
if trimmed_payload == "[DONE]" {
done = true;
break;
}
let payload: Value = serde_json::from_str(trimmed_payload).map_err(|err| {
StreamAssemblyError::InvalidPayload(err.to_string())
.into_llm_error("OpenAI")
})?;
let event_policy = response_stream_event_policy(&payload)
.map_err(|message| {
StreamAssemblyError::InvalidPayload(message.to_string())
.into_llm_error("OpenAI")
})?;
let event_type = payload
.get("type")
.and_then(Value::as_str)
.ok_or_else(|| {
StreamAssemblyError::MissingField("type").into_llm_error("OpenAI")
})?;
match event_policy {
ResponsesStreamEventPolicy::DocumentedStatusMarkerNoop
| ResponsesStreamEventPolicy::DocumentedValueBearingRigGap => {
}
ResponsesStreamEventPolicy::Unsupported => {
Err(StreamAssemblyError::InvalidPayload(format!(
"unsupported Responses stream event type `{event_type}`"
))
.into_llm_error("OpenAI"))?;
}
ResponsesStreamEventPolicy::MeaningfulConversion => match event_type {
"response.output_text.delta" => {
let delta = payload
.get("delta")
.and_then(|value| value.as_str())
.ok_or_else(|| {
StreamAssemblyError::MissingField("delta")
.into_llm_error("OpenAI")
})?;
telemetry.on_content_delta(delta);
for event in aggregator.handle_content(delta) {
yield event;
}
}
"response.refusal.delta" => {
let delta = payload
.get("delta")
.and_then(|value| value.as_str())
.ok_or_else(|| {
StreamAssemblyError::MissingField("delta")
.into_llm_error("OpenAI")
})?;
telemetry.on_content_delta(delta);
aggregator.content.push_str(delta);
}
"response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
let delta = payload
.get("delta")
.and_then(|value| value.as_str())
.ok_or_else(|| {
StreamAssemblyError::MissingField("delta")
.into_llm_error("OpenAI")
})?;
if retain_reasoning_summaries
&& let Some(delta) = aggregator.handle_reasoning(delta) {
telemetry.on_reasoning_delta(&delta);
yield provider::LLMStreamEvent::Reasoning { delta };
}
}
"response.reasoning_content.delta" => {
let delta = payload
.get("delta")
.and_then(|value| value.as_str())
.ok_or_else(|| {
StreamAssemblyError::MissingField("delta")
.into_llm_error("OpenAI")
})?;
if retain_reasoning_summaries
&& let Some(delta) = aggregator.handle_reasoning(delta) {
telemetry.on_reasoning_delta(&delta);
yield provider::LLMStreamEvent::Reasoning { delta };
}
}
"response.reasoning_text.done" => {
let text = optional_string_field(&payload, "text")?;
let delta = optional_string_field(&payload, "delta")?;
if retain_reasoning_summaries
&& let Some(text) = text.or(delta)
&& let Some(delta) = aggregator.handle_reasoning(&text) {
telemetry.on_reasoning_delta(&delta);
yield provider::LLMStreamEvent::Reasoning { delta };
}
}
"response.output_item.added" | "response.output_item.done" => {
if let Some(item) = payload.get("item") {
tool_call_state.capture_metadata(
&mut aggregator,
item,
payload
.get("output_index")
.and_then(Value::as_u64)
.map(|value| value as usize),
);
}
}
"response.function_call_arguments.delta" => {
tool_call_state.handle_arguments_delta(&mut aggregator, &payload)?;
telemetry.on_tool_call_delta();
}
"response.completed" => {
if let Some(response_value) = payload.get("response") {
final_response = Some(response_value.clone());
}
done = true;
}
"response.failed" | "response.incomplete" => {
let error_message = if let Some(err) = payload.get("response")
.and_then(|r| r.get("error"))
{
err.get("message")
.and_then(|v| v.as_str())
.unwrap_or("Unknown error")
} else {
"Unknown error from Responses API"
};
let formatted_error = error_display::format_llm_error("OpenAI", error_message);
Err(provider::LLMError::Provider {
message: formatted_error,
metadata: None,
})?;
}
"error" => {
let error_message = payload
.get("error")
.and_then(|error| error.get("message"))
.and_then(Value::as_str)
.unwrap_or("Unknown error from Responses API");
let formatted_error = error_display::format_llm_error("OpenAI", error_message);
Err(provider::LLMError::Provider {
message: formatted_error,
metadata: None,
})?;
}
_ => {
Err(StreamAssemblyError::InvalidPayload(format!(
"unhandled Responses stream event type `{event_type}`"
))
.into_llm_error("OpenAI"))?;
}
},
}
}
if done {
break;
}
}
if offset > 0 {
buf.drain(..offset);
offset = 0;
}
if done {
break;
}
}
let response_value = match final_response {
Some(value) => value,
None => {
let formatted_error = error_display::format_llm_error(
"OpenAI",
"Stream ended without a completion event",
);
Err(provider::LLMError::Provider { message: formatted_error, metadata: None })?
}
};
let final_aggregator_response = aggregator.finalize();
let mut response = match parse_responses_payload(response_value.clone(), model.clone(), include_metrics) {
Ok(response) => response,
Err(_)
if final_response_output_is_empty(&response_value)
&& streamed_response_is_usable(&final_aggregator_response) =>
{
let mut response = final_aggregator_response.clone();
merge_final_response_metadata(&mut response, &response_value, include_metrics);
response
}
Err(err) => Err(err)?,
};
if response.content.is_none() {
response.content = final_aggregator_response.content;
} else if let (Some(c), Some(agg_c)) = (&mut response.content, final_aggregator_response.content)
&& !c.contains(&agg_c) {
c.push_str(&agg_c);
}
if response.reasoning.is_none() {
response.reasoning = final_aggregator_response.reasoning;
}
if response.tool_calls.is_none() {
response.tool_calls = final_aggregator_response.tool_calls;
}
let response = strip_reasoning_for_model(&model, response);
yield provider::LLMStreamEvent::Completed { response: Box::new(response) };
};
Box::pin(stream)
}
fn optional_string_field(payload: &Value, field: &'static str) -> Result<Option<String>, provider::LLMError> {
match payload.get(field) {
Some(value) => value.as_str().map(|value| Some(value.to_string())).ok_or_else(|| {
StreamAssemblyError::InvalidPayload(format!("field `{field}` in stream payload must be a string"))
.into_llm_error("OpenAI")
}),
None => Ok(None),
}
}
#[cfg(test)]
mod tests {
use super::{
ChatStreamEvent, ChatToolCallState, ResponsesToolCallState, chat_event_to_normalized,
final_response_output_is_empty, merge_final_response_metadata, streamed_response_is_usable,
};
use crate::provider::{LLMResponse, NormalizedStreamEvent, ReasoningSource, ToolCall};
use crate::providers::shared::StreamAggregator;
use serde_json::json;
#[test]
fn responses_final_metadata_parses_cached_prompt_tokens_when_enabled() {
let mut response = LLMResponse::default();
merge_final_response_metadata(
&mut response,
&json!({
"id": "resp_stream",
"usage": {
"input_tokens": 12,
"output_tokens": 5,
"total_tokens": 17,
"input_tokens_details": {
"cached_tokens": 9
}
}
}),
true,
);
assert_eq!(response.request_id.as_deref(), Some("resp_stream"));
let usage = response.usage.expect("usage should be populated");
assert_eq!(usage.prompt_tokens, 12);
assert_eq!(usage.completion_tokens, 5);
assert_eq!(usage.total_tokens, 17);
assert_eq!(usage.cached_prompt_tokens, Some(9));
}
#[test]
fn empty_final_response_can_use_streamed_tool_call_delta() {
let response = LLMResponse {
tool_calls: Some(vec![ToolCall::function(
"call_1".to_string(),
"search_workspace".to_string(),
"{\"query\":\"vtcode\"}".to_string(),
)]),
..Default::default()
};
assert!(final_response_output_is_empty(&json!({"output": []})));
assert!(streamed_response_is_usable(&response));
}
#[test]
fn chat_normalization_preserves_reasoning_source() {
let raw_events = chat_event_to_normalized(ChatStreamEvent::ReasoningDelta {
delta: "tagged trace".to_string(),
source: ReasoningSource::Unknown,
});
assert!(matches!(
raw_events.as_slice(),
[NormalizedStreamEvent::ReasoningDelta { delta, source }]
if delta == "tagged trace" && *source == ReasoningSource::Unknown
));
let summary_events = chat_event_to_normalized(ChatStreamEvent::ReasoningDelta {
delta: "provider summary".to_string(),
source: ReasoningSource::ProviderSummary,
});
assert!(matches!(
summary_events.as_slice(),
[NormalizedStreamEvent::ReasoningDelta { delta, source }]
if delta == "provider summary" && *source == ReasoningSource::ProviderSummary
));
}
#[test]
fn chat_normalization_preserves_completed_usage_before_done() {
let events = chat_event_to_normalized(ChatStreamEvent::Completed {
response: Box::new(LLMResponse {
usage: Some(crate::provider::Usage {
prompt_tokens: 10,
completion_tokens: 5,
total_tokens: 15,
..Default::default()
}),
..Default::default()
}),
});
assert!(matches!(
events.as_slice(),
[NormalizedStreamEvent::Usage { .. }, NormalizedStreamEvent::Done { .. }]
));
}
#[test]
fn chat_tool_deltas_emit_structured_events_and_reuse_the_final_id() {
let mut aggregator = StreamAggregator::new("gpt-5".to_string());
let mut tool_call_state = ChatToolCallState::default();
let first = tool_call_state.handle_deltas(
&mut aggregator,
&[json!({
"index": 0,
"id": "call_1",
"function": {
"name": "search_workspace",
"arguments": "{\"query\":\"vt"
}
})],
);
assert!(matches!(
first.as_slice(),
[
ChatStreamEvent::ToolCallStart { call_id, name },
ChatStreamEvent::ToolCallDelta { call_id: delta_call_id, delta }
] if call_id == "call_1"
&& delta_call_id == "call_1"
&& name.as_deref() == Some("search_workspace")
&& delta == "{\"query\":\"vt"
));
let second = tool_call_state.handle_deltas(
&mut aggregator,
&[json!({
"index": 0,
"function": {
"arguments": "code\"}"
}
})],
);
assert!(matches!(
second.as_slice(),
[ChatStreamEvent::ToolCallDelta { call_id, delta }]
if call_id == "call_1" && delta == "code\"}"
));
let calls = aggregator.finalize().tool_calls.expect("tool call expected");
assert_eq!(calls[0].id, "call_1");
assert_eq!(calls[0].tool_name(), Some("search_workspace"));
assert_eq!(calls[0].raw_input(), Some("{\"query\":\"vtcode\"}"));
}
#[test]
fn responses_tool_call_state_uses_provider_call_id_when_item_id_differs() {
let mut aggregator = StreamAggregator::new("gpt-5".to_string());
let mut tool_call_state = ResponsesToolCallState::default();
tool_call_state.capture_metadata(
&mut aggregator,
&json!({
"type": "function_call",
"id": "fc_1",
"call_id": "call_1",
"name": "search_workspace"
}),
Some(0),
);
tool_call_state
.handle_arguments_delta(
&mut aggregator,
&json!({
"type": "response.function_call_arguments.delta",
"item_id": "fc_1",
"delta": "{\"query\":\"vtcode\"}"
}),
)
.expect("tool delta should parse");
let response = aggregator.finalize();
assert_eq!(
response.tool_calls.as_ref(),
Some(&vec![ToolCall::function(
"call_1".to_string(),
"search_workspace".to_string(),
"{\"query\":\"vtcode\"}".to_string(),
)])
);
}
#[test]
fn responses_tool_call_state_reuses_fabricated_id_across_deltas() {
let mut aggregator = StreamAggregator::new("gpt-5".to_string());
let mut tool_call_state = ResponsesToolCallState::default();
let delta = |fragment: &str| {
json!({
"type": "response.function_call_arguments.delta",
"output_index": 0,
"delta": fragment
})
};
tool_call_state
.handle_arguments_delta(&mut aggregator, &delta("{\"query\":"))
.expect("tool delta should parse");
tool_call_state
.handle_arguments_delta(&mut aggregator, &delta("\"vtcode\"}"))
.expect("tool delta should parse");
aggregator.handle_tool_calls(&[json!({
"index": 0,
"function": {"name": "search_workspace"}
})]);
let calls = aggregator.finalize().tool_calls.expect("tool call expected");
assert_eq!(calls.len(), 1, "both deltas must land on one builder");
assert!(calls[0].id.starts_with("call_"));
assert_ne!(calls[0].id, "call_0");
let function = calls[0].function.as_ref().expect("function expected");
assert_eq!(function.arguments, "{\"query\":\"vtcode\"}");
}
#[test]
fn responses_tool_call_state_fabricates_distinct_ids_per_decoder() {
let mut ids = Vec::new();
for _ in 0..2 {
let mut aggregator = StreamAggregator::new("gpt-5".to_string());
let mut tool_call_state = ResponsesToolCallState::default();
tool_call_state
.handle_arguments_delta(
&mut aggregator,
&json!({
"type": "response.function_call_arguments.delta",
"output_index": 0,
"delta": "{}"
}),
)
.expect("tool delta should parse");
aggregator.handle_tool_calls(&[json!({
"index": 0,
"function": {"name": "search_workspace"}
})]);
let calls = aggregator.finalize().tool_calls.expect("tool call expected");
ids.push(calls[0].id.clone());
}
assert_ne!(ids[0], ids[1], "fabricated ids must differ across responses");
}
}