vtcode_core/llm/providers/openai/
stream_decoder.rs1use crate::llm::error_display;
12use crate::llm::provider;
13use crate::llm::providers::shared::StreamTelemetry;
14use crate::llm::providers::shared::parse_cached_prompt_tokens_from_usage;
15use crate::llm::providers::shared::{StreamAssemblyError, extract_data_payload, find_sse_boundary};
16use crate::models_manager::model_family::find_family_for_model;
17use async_stream::try_stream;
18use futures::StreamExt;
19use serde_json::Value;
20use std::time::Instant;
21
22use super::responses_api::parse_responses_payload;
23use super::streaming::OpenAIStreamTelemetry;
24
25fn strip_reasoning_for_model(
26 model: &str,
27 mut response: provider::LLMResponse,
28) -> provider::LLMResponse {
29 if !find_family_for_model(model).supports_reasoning_summaries {
30 response.reasoning = None;
31 response.reasoning_details = None;
32 }
33
34 response
35}
36
37fn streamed_response_is_usable(response: &provider::LLMResponse) -> bool {
38 response
39 .content
40 .as_deref()
41 .is_some_and(|content| !content.is_empty())
42 || response
43 .tool_calls
44 .as_ref()
45 .is_some_and(|tool_calls| !tool_calls.is_empty())
46 || response
47 .reasoning
48 .as_deref()
49 .is_some_and(|reasoning| !reasoning.is_empty())
50 || response
51 .reasoning_details
52 .as_ref()
53 .is_some_and(|details| !details.is_empty())
54}
55
56fn final_response_output_is_empty(final_response: &Value) -> bool {
57 final_response
58 .get("output")
59 .and_then(Value::as_array)
60 .is_some_and(Vec::is_empty)
61}
62
63fn merge_final_response_metadata(
64 response: &mut provider::LLMResponse,
65 final_response: &Value,
66 include_cached_prompt_metrics: bool,
67) {
68 if let Some(usage_value) = final_response.get("usage") {
69 let cached_prompt_tokens =
70 parse_cached_prompt_tokens_from_usage(usage_value, include_cached_prompt_metrics);
71
72 response.usage = Some(provider::Usage {
73 prompt_tokens: usage_value
74 .get("input_tokens")
75 .or_else(|| usage_value.get("prompt_tokens"))
76 .and_then(Value::as_u64)
77 .and_then(|value| u32::try_from(value).ok())
78 .unwrap_or(0),
79 completion_tokens: usage_value
80 .get("output_tokens")
81 .or_else(|| usage_value.get("completion_tokens"))
82 .and_then(Value::as_u64)
83 .and_then(|value| u32::try_from(value).ok())
84 .unwrap_or(0),
85 total_tokens: usage_value
86 .get("total_tokens")
87 .and_then(Value::as_u64)
88 .and_then(|value| u32::try_from(value).ok())
89 .unwrap_or(0),
90 cached_prompt_tokens,
91 cache_creation_tokens: None,
92 cache_read_tokens: None,
93 iterations: None,
94 });
95 }
96
97 if let Some(request_id) = final_response
98 .get("id")
99 .and_then(Value::as_str)
100 .or_else(|| final_response.get("request_id").and_then(Value::as_str))
101 {
102 response.request_id = Some(request_id.to_string());
103 }
104}
105
106pub(crate) fn create_chat_stream(
107 response: reqwest::Response,
108 model: String,
109) -> provider::LLMStream {
110 let stream = try_stream! {
111 let mut body_stream = response.bytes_stream();
112 let mut buffer = String::new();
113 let retain_reasoning_summaries = find_family_for_model(&model).supports_reasoning_summaries;
114 let mut aggregator = crate::llm::providers::shared::StreamAggregator::new(model.clone());
115 let telemetry = OpenAIStreamTelemetry;
116
117 while let Some(chunk_result) = body_stream.next().await {
118 let chunk = chunk_result.map_err(|err| {
119 let formatted_error = error_display::format_llm_error(
120 "OpenAI",
121 &format!("Streaming error: {}", err),
122 );
123 provider::LLMError::Network { message: formatted_error, metadata: None }
124 })?;
125
126 buffer.push_str(&String::from_utf8_lossy(&chunk));
127
128 while let Some((split_idx, delimiter_len)) = find_sse_boundary(&buffer) {
129 let event = buffer[..split_idx].to_string();
130 buffer.drain(..split_idx + delimiter_len);
131
132 if let Some(data_payload) = extract_data_payload(&event) {
133 let trimmed_payload = data_payload.trim();
134 if trimmed_payload.is_empty() || trimmed_payload == "[DONE]" {
135 continue;
136 }
137
138 let payload: Value = serde_json::from_str(trimmed_payload).map_err(|err| {
139 StreamAssemblyError::InvalidPayload(err.to_string())
140 .into_llm_error("OpenAI")
141 })?;
142
143 if let Some(usage_val) = payload.get("usage")
144 && let Ok(u) = serde_json::from_value::<provider::Usage>(usage_val.clone()) {
145 aggregator.set_usage(u);
146 }
147
148 if let Some(choices) = payload.get("choices").and_then(|v| v.as_array())
149 && let Some(choice) = choices.first() {
150 if let Some(delta) = choice.get("delta") {
151 if let Some(content) = delta.get("content").and_then(|v| v.as_str()) {
152 telemetry.on_content_delta(content);
153 for event in aggregator.handle_content(content) {
154 yield event;
155 }
156 }
157
158 if retain_reasoning_summaries
159 && let Some(reasoning) = delta.get("reasoning_content").and_then(|v| v.as_str())
160 && let Some(delta) = aggregator.handle_reasoning(reasoning) {
161 telemetry.on_reasoning_delta(&delta);
162 yield provider::LLMStreamEvent::Reasoning { delta };
163 }
164
165 if let Some(tool_deltas) = delta.get("tool_calls").and_then(|v| v.as_array()) {
166 aggregator.handle_tool_calls(tool_deltas);
167 telemetry.on_tool_call_delta();
168 }
169 }
170
171 if let Some(reason) = choice.get("finish_reason").and_then(|v| v.as_str()) {
172 aggregator.set_finish_reason(match reason {
173 "stop" => provider::FinishReason::Stop,
174 "length" => provider::FinishReason::Length,
175 "tool_calls" => provider::FinishReason::ToolCalls,
176 "content_filter" => provider::FinishReason::ContentFilter,
177 _ => provider::FinishReason::Stop,
178 });
179 }
180 }
181 }
182 }
183 }
184
185 let response = aggregator.finalize();
186 let response = strip_reasoning_for_model(&model, response);
187 yield provider::LLMStreamEvent::Completed { response: Box::new(response) };
188 };
189
190 Box::pin(stream)
191}
192
193#[cfg(test)]
194mod tests {
195 use super::{
196 final_response_output_is_empty, merge_final_response_metadata, streamed_response_is_usable,
197 };
198 use crate::llm::provider::{LLMResponse, ToolCall};
199 use serde_json::json;
200
201 #[test]
202 fn responses_final_metadata_parses_cached_prompt_tokens_when_enabled() {
203 let mut response = LLMResponse::default();
204 merge_final_response_metadata(
205 &mut response,
206 &json!({
207 "id": "resp_stream",
208 "usage": {
209 "input_tokens": 12,
210 "output_tokens": 5,
211 "total_tokens": 17,
212 "input_tokens_details": {
213 "cached_tokens": 9
214 }
215 }
216 }),
217 true,
218 );
219
220 assert_eq!(response.request_id.as_deref(), Some("resp_stream"));
221 let usage = response.usage.expect("usage should be populated");
222 assert_eq!(usage.prompt_tokens, 12);
223 assert_eq!(usage.completion_tokens, 5);
224 assert_eq!(usage.total_tokens, 17);
225 assert_eq!(usage.cached_prompt_tokens, Some(9));
226 }
227
228 #[test]
229 fn empty_final_response_can_use_streamed_tool_call_delta() {
230 let response = LLMResponse {
231 tool_calls: Some(vec![ToolCall::function(
232 "call_1".to_string(),
233 "search_workspace".to_string(),
234 "{\"query\":\"vtcode\"}".to_string(),
235 )]),
236 ..Default::default()
237 };
238
239 assert!(final_response_output_is_empty(&json!({"output": []})));
240 assert!(streamed_response_is_usable(&response));
241 }
242}
243
244pub(crate) fn create_responses_stream(
245 response: reqwest::Response,
246 model: String,
247 include_metrics: bool,
248 _debug_model: Option<String>,
249 _request_timer: Option<Instant>,
250) -> provider::LLMStream {
251 let stream = try_stream! {
252 let mut body_stream = response.bytes_stream();
253 let mut buffer = String::new();
254 let mut aggregator = crate::llm::providers::shared::StreamAggregator::new(model.clone());
255 let retain_reasoning_summaries = find_family_for_model(&model).supports_reasoning_summaries;
256 let mut final_response: Option<Value> = None;
257 let mut done = false;
258 #[cfg(debug_assertions)]
259 let mut streamed_events_counter: usize = 0;
260 let telemetry = OpenAIStreamTelemetry;
261
262 while let Some(chunk_result) = body_stream.next().await {
263 let chunk = chunk_result.map_err(|err| {
264 let formatted_error = error_display::format_llm_error(
265 "OpenAI",
266 &format!("Streaming error: {}", err),
267 );
268 provider::LLMError::Network { message: formatted_error, metadata: None }
269 })?;
270
271 buffer.push_str(&String::from_utf8_lossy(&chunk));
272
273 while let Some((split_idx, delimiter_len)) = find_sse_boundary(&buffer) {
274 let event = buffer[..split_idx].to_string();
275 buffer.drain(..split_idx + delimiter_len);
276 #[cfg(debug_assertions)]
277 {
278 streamed_events_counter = streamed_events_counter.saturating_add(1);
279 }
280
281 if let Some(data_payload) = extract_data_payload(&event) {
282 let trimmed_payload = data_payload.trim();
283 if trimmed_payload.is_empty() {
284 continue;
285 }
286
287 if trimmed_payload == "[DONE]" {
288 done = true;
289 break;
290 }
291
292 let payload: Value = serde_json::from_str(trimmed_payload).map_err(|err| {
293 StreamAssemblyError::InvalidPayload(err.to_string())
294 .into_llm_error("OpenAI")
295 })?;
296
297 if let Some(event_type) = payload.get("type").and_then(|value| value.as_str()) {
298 match event_type {
299 "response.output_text.delta" => {
300 let delta = payload
301 .get("delta")
302 .and_then(|value| value.as_str())
303 .ok_or_else(|| {
304 StreamAssemblyError::MissingField("delta")
305 .into_llm_error("OpenAI")
306 })?;
307 telemetry.on_content_delta(delta);
308
309 for event in aggregator.handle_content(delta) {
310 yield event;
311 }
312 }
313 "response.refusal.delta" => {
314 let delta = payload
315 .get("delta")
316 .and_then(|value| value.as_str())
317 .ok_or_else(|| {
318 StreamAssemblyError::MissingField("delta")
319 .into_llm_error("OpenAI")
320 })?;
321 telemetry.on_content_delta(delta);
322 aggregator.content.push_str(delta);
323 }
324 "response.reasoning_text.delta" | "response.reasoning_summary_text.delta" => {
325 let delta = payload
326 .get("delta")
327 .and_then(|value| value.as_str())
328 .ok_or_else(|| {
329 StreamAssemblyError::MissingField("delta")
330 .into_llm_error("OpenAI")
331 })?;
332 if retain_reasoning_summaries
333 && let Some(delta) = aggregator.handle_reasoning(delta) {
334 telemetry.on_reasoning_delta(&delta);
335 yield provider::LLMStreamEvent::Reasoning { delta };
336 }
337 }
338 "response.function_call_arguments.delta" => {}
339 "response.completed" => {
340 if let Some(response_value) = payload.get("response") {
341 final_response = Some(response_value.clone());
342 }
343 done = true;
344 }
345 "response.failed" | "response.incomplete" => {
346 let error_message = if let Some(err) = payload.get("response")
347 .and_then(|r| r.get("error"))
348 {
349 err.get("message")
350 .and_then(|v| v.as_str())
351 .unwrap_or("Unknown error")
352 } else {
353 "Unknown error from Responses API"
354 };
355 let formatted_error = error_display::format_llm_error("OpenAI", error_message);
356 Err(provider::LLMError::Provider {
357 message: formatted_error,
358 metadata: None,
359 })?;
360 }
361 _ => {}
362 }
363 }
364 }
365
366 if done {
367 break;
368 }
369 }
370
371 if done {
372 break;
373 }
374 }
375
376 let response_value = match final_response {
377 Some(value) => value,
378 None => {
379 let formatted_error = error_display::format_llm_error(
380 "OpenAI",
381 "Stream ended without a completion event",
382 );
383 Err(provider::LLMError::Provider { message: formatted_error, metadata: None })?
384 }
385 };
386
387 let final_aggregator_response = aggregator.finalize();
388 let mut response = match parse_responses_payload(response_value.clone(), model.clone(), include_metrics) {
389 Ok(response) => response,
390 Err(_)
391 if final_response_output_is_empty(&response_value)
392 && streamed_response_is_usable(&final_aggregator_response) =>
393 {
394 let mut response = final_aggregator_response.clone();
395 merge_final_response_metadata(&mut response, &response_value, include_metrics);
396 response
397 }
398 Err(err) => Err(err)?,
399 };
400
401 if response.content.is_none() {
402 response.content = final_aggregator_response.content;
403 } else if let (Some(c), Some(agg_c)) = (&mut response.content, final_aggregator_response.content)
404 && !c.contains(&agg_c) {
405 c.push_str(&agg_c);
406 }
407
408 if response.reasoning.is_none() {
409 response.reasoning = final_aggregator_response.reasoning;
410 }
411
412 let response = strip_reasoning_for_model(&model, response);
413 yield provider::LLMStreamEvent::Completed { response: Box::new(response) };
414 };
415
416 Box::pin(stream)
417}