magi_code/providers/
stream.rs

1use crate::providers::{
2    error::ProviderError,
3    sse::{diagnostic_snippet, next_sse_event_boundary, sse_data},
4};
5use serde::{Deserialize, Serialize};
6use serde_json::Value;
7use std::collections::{BTreeMap, BTreeSet, btree_map::Entry};
8
9mod chat_completions;
10mod inline_tools;
11mod openai_responses;
12mod reasoning;
13
14use chat_completions::chat_finish_reason;
15use inline_tools::normalize_extra_quoted_tool_arguments;
16use openai_responses::{
17    is_whole_response_completion, is_whole_response_failure, whole_response_failure_detail,
18};
19use reasoning::reasoning_summary_text;
20
21#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
22pub struct ToolCall {
23    pub id: String,
24    pub name: String,
25    pub arguments: Value,
26}
27
28#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
29pub struct Usage {
30    pub input: u64,
31    pub output: u64,
32    pub cache_read: u64,
33    pub cache_write: u64,
34    pub total: u64,
35    pub reasoning_tokens: Option<u64>,
36}
37
38#[derive(Debug, Clone, Copy, Default, Serialize, Deserialize, PartialEq, Eq)]
39pub(crate) struct UsagePresence {
40    pub(crate) input: bool,
41    pub(crate) output: bool,
42    pub(crate) cache_read: bool,
43    pub(crate) cache_write: bool,
44    pub(crate) total: bool,
45    pub(crate) reasoning: bool,
46}
47
48#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
49pub(crate) struct UsageObservation {
50    pub(crate) usage: Usage,
51    pub(crate) presence: UsagePresence,
52}
53
54#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
55pub struct ReasoningSummary {
56    pub text: String,
57    pub item_id: Option<String>,
58    pub turn_id: Option<String>,
59    /// True only for explicit provider summary fields, never raw reasoning or inline thinking.
60    #[serde(default)]
61    pub provider_summary: bool,
62}
63
64const MAX_SSE_EVENT_BUFFER_BYTES: usize = 1024 * 1024;
65const MAX_TOOL_ARGUMENT_BYTES: usize = 1024 * 1024;
66
67#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
68pub enum ProviderEvent {
69    TextDelta(String),
70    ReasoningSummaryDelta(String),
71    ReasoningSummaryComplete(String),
72    ReasoningSummaryCompleteIdentified(ReasoningSummary),
73    ToolCall(ToolCall),
74    /// Complete provider call rejected before dispatch; return its error as a tool result.
75    ToolCallRejected {
76        call: ToolCall,
77        reason: String,
78    },
79    ResponseItem(Value),
80    Usage(Usage),
81    UsagePartial(Usage),
82    UsageObserved(UsageObservation),
83    /// Local-only metadata observation; never persist or replay this event.
84    ServiceTier(String),
85    Done,
86    ResponseIdentity(crate::providers::error::ResponseAttemptIdentity),
87}
88pub(crate) fn is_semantic_progress_event(event: &ProviderEvent) -> bool {
89    match event {
90        ProviderEvent::TextDelta(_)
91        | ProviderEvent::ReasoningSummaryDelta(_)
92        | ProviderEvent::ReasoningSummaryComplete(_)
93        | ProviderEvent::ReasoningSummaryCompleteIdentified(_)
94        | ProviderEvent::ToolCall(_)
95        | ProviderEvent::ToolCallRejected { .. }
96        | ProviderEvent::Usage(_)
97        | ProviderEvent::UsagePartial(_)
98        | ProviderEvent::UsageObserved(_)
99        | ProviderEvent::Done => true,
100        ProviderEvent::ResponseItem(item) => {
101            matches!(
102                item.get("type").and_then(Value::as_str),
103                Some(
104                    "function_call"
105                        | "function_call_output"
106                        | "reasoning"
107                        | "thinking"
108                        | "redacted_thinking"
109                )
110            ) || item
111                .get("tool_calls")
112                .and_then(Value::as_array)
113                .is_some_and(|tool_calls| !tool_calls.is_empty())
114        }
115        ProviderEvent::ServiceTier(_) | ProviderEvent::ResponseIdentity(_) => false,
116    }
117}
118#[derive(Debug, Clone, Default, PartialEq)]
119pub(crate) struct StreamParseOutcome {
120    pub(crate) events: Vec<ProviderEvent>,
121    pub(crate) semantic_progress: bool,
122    pub(crate) unsafe_recovery_progress: bool,
123}
124
125#[derive(Debug, Default)]
126pub(crate) struct StreamParser {
127    event_buffer: String,
128    tool_calls: BTreeMap<String, PendingToolCall>,
129    chat_tool_call_indices: BTreeMap<String, String>,
130    emitted_response_item_keys: BTreeSet<String>,
131    reasoning_summary_text: String,
132    saw_raw_reasoning: bool,
133    completed_reasoning_summary_text: Option<String>,
134    completed_reasoning_summary_keys: BTreeMap<String, String>,
135    saw_identified_reasoning_completion: bool,
136    emitted_text_delta: bool,
137    saw_terminal_completion: bool,
138    emitted_done: bool,
139    unsafe_chat_tool_call_completion: bool,
140    next_tool_call_sequence: u64,
141    content_buffer: String,
142    content_search_start: usize,
143    thinking_complete: bool,
144    gemma_inline_tool_calls_enabled: bool,
145    gemma_inline_tool_call_counter: u64,
146    response_model: Option<String>,
147    returned_service_tier: Option<String>,
148}
149#[derive(Debug, Clone, Default, PartialEq, Eq)]
150struct PendingToolCall {
151    call_id: Option<String>,
152    name: Option<String>,
153    arguments_text: String,
154    emitted: bool,
155    provider_index: Option<u64>,
156    first_seen_sequence: u64,
157    source: ToolCallSource,
158}
159
160#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
161enum ToolCallSource {
162    #[default]
163    Responses,
164    ChatCompletions,
165}
166
167fn gemma_inline_tool_calls_enabled(provider_id: &str, model: &str) -> bool {
168    let provider_id = provider_id.to_ascii_lowercase();
169    let model = model.to_ascii_lowercase();
170    let custom_vllm_profile = provider_id.contains("vllm") || provider_id.contains("foundry");
171    custom_vllm_profile && (model.contains("gemma") || model.contains("diffusiongemma"))
172}
173
174impl StreamParser {
175    pub(crate) fn for_provider_model(provider_id: &str, model: &str) -> Self {
176        Self {
177            gemma_inline_tool_calls_enabled: gemma_inline_tool_calls_enabled(provider_id, model),
178            ..Self::default()
179        }
180    }
181
182    #[cfg(test)]
183    pub(crate) fn push_chunk(&mut self, chunk: &str) -> anyhow::Result<Vec<ProviderEvent>> {
184        Ok(self.push_chunk_outcome(chunk)?.events)
185    }
186
187    pub(crate) fn push_chunk_outcome(&mut self, chunk: &str) -> anyhow::Result<StreamParseOutcome> {
188        self.event_buffer.push_str(chunk);
189        let mut buffer = std::mem::take(&mut self.event_buffer);
190        let mut events = Vec::new();
191        let mut semantic_progress = false;
192        let mut unsafe_recovery_progress = false;
193        while let Some((boundary, boundary_len)) = next_sse_event_boundary(&buffer) {
194            let parsed = sse_data(&buffer[..boundary]).map(|data| self.parse_data_event(&data));
195            buffer.drain(..boundary + boundary_len);
196            if let Some(parsed) = parsed {
197                let parsed = match parsed {
198                    Ok(parsed) => parsed,
199                    Err(error) => {
200                        self.event_buffer = buffer;
201                        return Err(error);
202                    }
203                };
204                semantic_progress |= parsed.semantic_progress
205                    || parsed.events.iter().any(is_semantic_progress_event);
206                unsafe_recovery_progress |= parsed.unsafe_recovery_progress;
207                events.extend(parsed.events);
208            }
209        }
210        self.event_buffer = buffer;
211        self.ensure_event_buffer_within_limit()?;
212        Ok(StreamParseOutcome {
213            events,
214            semantic_progress,
215            unsafe_recovery_progress,
216        })
217    }
218
219    pub(crate) fn finish(&mut self) -> anyhow::Result<Vec<ProviderEvent>> {
220        if !self.event_buffer.trim().is_empty() {
221            let raw_event = std::mem::take(&mut self.event_buffer);
222            if let Some(data) = sse_data(&raw_event)
223                && data == "[DONE]"
224            {
225                self.saw_terminal_completion = true;
226                return self.done_event(true);
227            }
228            anyhow::bail!(
229                "provider SSE stream ended with incomplete event buffer: {}",
230                diagnostic_snippet(&raw_event)
231            );
232        }
233        for (key, pending) in &self.tool_calls {
234            if !pending.emitted && !pending.arguments_text.trim().is_empty() {
235                parse_arguments_text(&pending.arguments_text).map_err(|error| {
236                    anyhow::anyhow!(
237                        "provider SSE stream ended with incomplete tool call arguments for {key}: {error}: {}",
238                        diagnostic_snippet(&pending.arguments_text)
239                    )
240                })?;
241            }
242        }
243        if !self.saw_terminal_completion {
244            return Err(ProviderError::stream_terminal(
245                "missing provider stream completion before EOF",
246            )
247            .into());
248        }
249        Ok(Vec::new())
250    }
251
252    fn parse_data_event(&mut self, data: &str) -> anyhow::Result<StreamParseOutcome> {
253        let mut events = Vec::new();
254        let mut semantic_progress = false;
255        if data == "[DONE]" {
256            semantic_progress |= !self.saw_terminal_completion;
257            self.saw_terminal_completion = true;
258            events.extend(self.done_event(true)?);
259            return Ok(StreamParseOutcome {
260                events,
261                semantic_progress,
262                unsafe_recovery_progress: false,
263            });
264        }
265        let value = serde_json::from_str::<Value>(data).map_err(|error| {
266            anyhow::anyhow!(
267                "malformed provider SSE data JSON: {error}: {}",
268                diagnostic_snippet(data)
269            )
270        })?;
271        let item_type = value
272            .get("type")
273            .and_then(Value::as_str)
274            .unwrap_or_default();
275        self.response_model = self.response_model.clone().or_else(|| {
276            value
277                .pointer("/response/model")
278                .and_then(Value::as_str)
279                .map(crate::providers::error::bounded_response_identity_string)
280                .or_else(|| {
281                    value
282                        .get("model")
283                        .and_then(Value::as_str)
284                        .map(crate::providers::error::bounded_response_identity_string)
285                })
286        });
287        // Metadata is local-only and deduplicated per response.
288        let returned_tier = value
289            .get("service_tier")
290            .or_else(|| value.pointer("/response/service_tier"))
291            .and_then(crate::fast::parse_returned_service_tier);
292        if let Some(tier) = returned_tier
293            && self.returned_service_tier.as_deref() != Some(tier.as_str())
294        {
295            self.returned_service_tier = Some(tier.clone());
296            events.push(ProviderEvent::ServiceTier(tier));
297        }
298        if is_whole_response_failure(&value, item_type) {
299            let message = match whole_response_failure_detail(&value) {
300                Some(detail) => format!(
301                    "provider stream ended with failed or incomplete response: {detail} | raw: {}",
302                    diagnostic_snippet(data)
303                ),
304                None => format!(
305                    "provider stream ended with failed or incomplete response: {}",
306                    diagnostic_snippet(data)
307                ),
308            };
309            return Err(ProviderError::stream_failed_incomplete(message).into());
310        }
311        let chat_finish_reason = chat_finish_reason(&value);
312        if matches!(
313            chat_finish_reason,
314            Some(reason) if !matches!(reason, "stop" | "tool_calls")
315        ) {
316            self.unsafe_chat_tool_call_completion = true;
317        }
318        let chat_finish_is_terminal = matches!(
319            chat_finish_reason,
320            Some("stop" | "tool_calls" | "length" | "content_filter")
321        );
322        let is_terminal_completion =
323            is_whole_response_completion(&value, item_type) || chat_finish_is_terminal;
324        if let Some((summary, provider_summary)) =
325            self.reasoning_summary_delta_from_event(&value, item_type)
326        {
327            self.reasoning_summary_text.push_str(summary);
328            self.saw_raw_reasoning |= !provider_summary;
329            semantic_progress = true;
330            events.push(ProviderEvent::ReasoningSummaryDelta(summary.to_string()));
331        }
332        if let Some((summary, item_id)) = self.reasoning_summary_done_from_event(&value, item_type)
333        {
334            events.extend(self.reconcile_reasoning_summary_complete(summary, item_id));
335        }
336        if !matches!(
337            item_type,
338            "response.function_call_arguments.delta"
339                | "response.reasoning_summary_text.delta"
340                | "response.reasoning_summary_text.done"
341        ) {
342            if let Some(delta) = self.chat_content_delta_from_event(&value) {
343                let (reasoning_delta, text_delta) = self.process_chat_content_delta(delta);
344                if let Some(reasoning_delta) = reasoning_delta {
345                    self.reasoning_summary_text.push_str(&reasoning_delta);
346                    self.saw_raw_reasoning = true;
347                    semantic_progress = true;
348                    events.push(ProviderEvent::ReasoningSummaryDelta(reasoning_delta));
349                }
350                if let Some(text_delta) = text_delta {
351                    self.emitted_text_delta = true;
352                    semantic_progress = true;
353                    events.push(ProviderEvent::TextDelta(text_delta));
354                }
355            } else if let Some(delta) = self.text_delta_from_event(&value, item_type) {
356                self.emitted_text_delta = true;
357                semantic_progress = true;
358                events.push(ProviderEvent::TextDelta(delta.to_string()));
359            }
360        }
361        let response_items = self.parse_response_items(&value, item_type);
362        for item in response_items {
363            if let Some(summary) = reasoning_summary_text(&item) {
364                events.extend(self.reconcile_reasoning_summary_complete(
365                    &summary,
366                    item.get("id").and_then(Value::as_str),
367                ));
368            }
369            events.push(ProviderEvent::ResponseItem(item));
370        }
371        let (tool_calls, tool_call_progress) = self.parse_tool_calls(&value)?;
372        semantic_progress |= tool_call_progress;
373        if matches!(chat_finish_reason, Some("tool_calls" | "stop")) {
374            self.flush_pending_chat_content(&mut events);
375            self.emit_completed_chat_tool_calls(&mut events)?;
376        }
377        events.extend(tool_calls.into_iter().map(ProviderEvent::ToolCall));
378        if let Some(parsed_usage) = parse_usage(&value) {
379            events.push(ProviderEvent::UsageObserved(UsageObservation {
380                usage: parsed_usage.usage,
381                presence: parsed_usage.presence,
382            }));
383        }
384        if is_terminal_completion {
385            self.flush_pending_chat_content(&mut events);
386            self.saw_terminal_completion = true;
387            semantic_progress = true;
388            events.extend(self.done_event(false)?);
389        }
390        semantic_progress |= events.iter().any(is_semantic_progress_event);
391        Ok(StreamParseOutcome {
392            events,
393            semantic_progress,
394            unsafe_recovery_progress: tool_call_progress,
395        })
396    }
397
398    fn done_event(&mut self, complete_chat_tool_calls: bool) -> anyhow::Result<Vec<ProviderEvent>> {
399        let mut events = Vec::new();
400        self.flush_pending_chat_content(&mut events);
401        if complete_chat_tool_calls && !self.unsafe_chat_tool_call_completion {
402            self.emit_completed_chat_tool_calls(&mut events)?;
403        }
404        if !self.saw_identified_reasoning_completion
405            && !self.reasoning_summary_text.trim().is_empty()
406            && self.completed_reasoning_summary_text.as_deref()
407                != Some(self.reasoning_summary_text.as_str())
408        {
409            self.completed_reasoning_summary_text = Some(self.reasoning_summary_text.clone());
410            if self.saw_raw_reasoning {
411                events.push(ProviderEvent::ReasoningSummaryComplete(
412                    self.reasoning_summary_text.clone(),
413                ));
414            } else {
415                events.push(ProviderEvent::ReasoningSummaryCompleteIdentified(
416                    ReasoningSummary {
417                        text: self.reasoning_summary_text.clone(),
418                        item_id: None,
419                        turn_id: None,
420                        provider_summary: true,
421                    },
422                ));
423            }
424        }
425        if !self.emitted_done {
426            self.emitted_done = true;
427            events.push(ProviderEvent::Done);
428        }
429        Ok(events)
430    }
431
432    fn parse_tool_calls(&mut self, value: &Value) -> anyhow::Result<(Vec<ToolCall>, bool)> {
433        let mut calls = Vec::new();
434        let mut semantic_progress = false;
435        semantic_progress |= self.parse_response_tool_calls(value, &mut calls)?;
436        semantic_progress |= self.parse_chat_tool_calls(value, &mut calls)?;
437        Ok((calls, semantic_progress))
438    }
439
440    fn pending_for_key(
441        &mut self,
442        key: &str,
443        provider_index: Option<u64>,
444        source: ToolCallSource,
445    ) -> &mut PendingToolCall {
446        match self.tool_calls.entry(key.to_string()) {
447            Entry::Occupied(entry) => {
448                let pending = entry.into_mut();
449                if pending.provider_index.is_none() {
450                    pending.provider_index = provider_index;
451                }
452                if pending.source != source {
453                    pending.source = source;
454                }
455                pending
456            }
457            Entry::Vacant(entry) => {
458                let sequence = self.next_tool_call_sequence;
459                self.next_tool_call_sequence += 1;
460                entry.insert(PendingToolCall {
461                    provider_index,
462                    first_seen_sequence: sequence,
463                    source,
464                    ..PendingToolCall::default()
465                })
466            }
467        }
468    }
469
470    fn migrate_pending_tool_call(&mut self, from_key: &str, to_key: &str) -> anyhow::Result<bool> {
471        if from_key == to_key {
472            return Ok(false);
473        }
474        let Some(from_pending) = self.tool_calls.remove(from_key) else {
475            return Ok(false);
476        };
477        match self.tool_calls.entry(to_key.to_string()) {
478            Entry::Vacant(entry) => {
479                entry.insert(from_pending);
480            }
481            Entry::Occupied(mut entry) => {
482                merge_pending_tool_call(entry.get_mut(), from_pending, to_key)?;
483            }
484        }
485        Ok(true)
486    }
487
488    fn ensure_event_buffer_within_limit(&self) -> anyhow::Result<()> {
489        if self.event_buffer.len() <= MAX_SSE_EVENT_BUFFER_BYTES {
490            return Ok(());
491        }
492        Err(ProviderError::stream_terminal(format!(
493            "provider SSE event exceeded maximum buffered size of {MAX_SSE_EVENT_BUFFER_BYTES} bytes before a frame boundary"
494        ))
495        .into())
496    }
497
498    fn push_tool_arguments_delta(
499        pending: &mut PendingToolCall,
500        delta: &str,
501    ) -> anyhow::Result<bool> {
502        if delta.is_empty() {
503            return Ok(false);
504        }
505        let next_len = pending.arguments_text.len().saturating_add(delta.len());
506        if next_len > MAX_TOOL_ARGUMENT_BYTES {
507            return Err(ProviderError::stream_terminal(format!(
508                "provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
509            ))
510            .into());
511        }
512        pending.arguments_text.push_str(delta);
513        Ok(true)
514    }
515
516    fn set_tool_arguments_text(
517        pending: &mut PendingToolCall,
518        arguments_text: String,
519    ) -> anyhow::Result<bool> {
520        if arguments_text.len() > MAX_TOOL_ARGUMENT_BYTES {
521            return Err(ProviderError::stream_terminal(format!(
522                "provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
523            ))
524            .into());
525        }
526        if pending.arguments_text == arguments_text {
527            return Ok(false);
528        }
529        pending.arguments_text = arguments_text;
530        Ok(true)
531    }
532    pub(crate) fn response_model(&self) -> Option<String> {
533        self.response_model.clone()
534    }
535}
536
537fn merge_pending_tool_call(
538    target: &mut PendingToolCall,
539    source: PendingToolCall,
540    key: &str,
541) -> anyhow::Result<()> {
542    if let Some(call_id) = source.call_id {
543        if let Some(existing) = &target.call_id
544            && existing != &call_id
545        {
546            anyhow::bail!(
547                "conflicting duplicate provider tool call id for {key}: {existing} vs {call_id}"
548            );
549        }
550        target.call_id = Some(call_id);
551    }
552    if let Some(name) = source.name {
553        if let Some(existing) = &target.name
554            && existing != &name
555        {
556            anyhow::bail!(
557                "conflicting duplicate provider tool call name for {key}: {existing} vs {name}"
558            );
559        }
560        target.name = Some(name);
561    }
562    if !source.arguments_text.is_empty() {
563        let target_arguments = std::mem::take(&mut target.arguments_text);
564        target.arguments_text = source.arguments_text;
565        let next_len = target
566            .arguments_text
567            .len()
568            .saturating_add(target_arguments.len());
569        if next_len > MAX_TOOL_ARGUMENT_BYTES {
570            return Err(ProviderError::stream_terminal(format!(
571                "provider tool call arguments exceeded maximum size of {MAX_TOOL_ARGUMENT_BYTES} bytes"
572            ))
573            .into());
574        }
575        target.arguments_text.push_str(&target_arguments);
576    }
577    target.emitted |= source.emitted;
578    target.provider_index = target.provider_index.or(source.provider_index);
579    target.first_seen_sequence = target.first_seen_sequence.min(source.first_seen_sequence);
580    target.source = source_priority(target.source, source.source);
581    Ok(())
582}
583
584fn source_priority(left: ToolCallSource, right: ToolCallSource) -> ToolCallSource {
585    if matches!(left, ToolCallSource::ChatCompletions)
586        || matches!(right, ToolCallSource::ChatCompletions)
587    {
588        ToolCallSource::ChatCompletions
589    } else {
590        ToolCallSource::Responses
591    }
592}
593
594fn arguments_as_text(arguments: &Value) -> String {
595    match arguments {
596        Value::String(text) => text.clone(),
597        value => value.to_string(),
598    }
599}
600
601fn parse_arguments_text(text: &str) -> anyhow::Result<Value> {
602    if text.trim().is_empty() {
603        return Ok(Value::Object(Default::default()));
604    }
605    parse_arguments_json_value(text)
606        .map(normalize_extra_quoted_tool_arguments)
607        .map_err(|error| {
608            anyhow::anyhow!(
609                "malformed non-empty provider tool call arguments: {error}: {}",
610                diagnostic_snippet(text)
611            )
612        })
613}
614
615fn parse_arguments_json_value(text: &str) -> serde_json::Result<Value> {
616    match serde_json::from_str::<Value>(text) {
617        Ok(value) => Ok(value),
618        Err(strict_error) => {
619            let mut stream = serde_json::Deserializer::from_str(text).into_iter::<Value>();
620            let value = match stream.next() {
621                Some(Ok(value)) => value,
622                Some(Err(error)) => return Err(error),
623                None => return Err(strict_error),
624            };
625            let trailing = text[stream.byte_offset()..].trim();
626            if trailing.is_empty() || trailing_is_empty_json_objects(trailing) {
627                Ok(value)
628            } else {
629                Err(strict_error)
630            }
631        }
632    }
633}
634
635fn trailing_is_empty_json_objects(mut text: &str) -> bool {
636    loop {
637        text = text.trim_start();
638        if text.is_empty() {
639            return true;
640        }
641        let Some(rest) = text.strip_prefix("{}") else {
642            return false;
643        };
644        text = rest;
645    }
646}
647
648#[derive(Debug, Clone, Default, PartialEq, Eq)]
649pub(crate) struct ParsedUsage {
650    pub(crate) usage: Usage,
651    pub(crate) input_tokens: Option<u64>,
652    pub(crate) reasoning_tokens: Option<u64>,
653    pub(crate) presence: UsagePresence,
654}
655
656fn parse_usage(value: &Value) -> Option<ParsedUsage> {
657    let usage = value
658        .pointer("/usage")
659        .filter(|usage| usage.is_object())
660        .or_else(|| value.pointer("/response/usage"))?
661        .as_object()?;
662    let input_tokens = usage
663        .get("input_tokens")
664        .and_then(Value::as_u64)
665        .or_else(|| usage.get("prompt_tokens").and_then(Value::as_u64));
666    let output_tokens = usage
667        .get("output_tokens")
668        .and_then(Value::as_u64)
669        .or_else(|| usage.get("completion_tokens").and_then(Value::as_u64));
670    let cache_read_tokens = usage
671        .get("input_tokens_details")
672        .and_then(|d| d.get("cached_tokens"))
673        .and_then(Value::as_u64)
674        .or_else(|| {
675            usage
676                .get("prompt_tokens_details")
677                .and_then(|d| d.get("cached_tokens"))
678                .and_then(Value::as_u64)
679        });
680    let cache_write_tokens = usage
681        .get("cache_write_tokens")
682        .and_then(Value::as_u64)
683        .or_else(|| {
684            usage
685                .get("input_tokens_details")
686                .and_then(|d| d.get("cache_write_tokens"))
687                .and_then(Value::as_u64)
688        })
689        .or_else(|| {
690            usage
691                .get("prompt_tokens_details")
692                .and_then(|d| d.get("cache_write_tokens"))
693                .and_then(Value::as_u64)
694        });
695    let total_tokens = usage.get("total_tokens").and_then(Value::as_u64);
696    let reasoning_tokens = usage
697        .get("output_tokens_details")
698        .and_then(|d| d.get("reasoning_tokens"))
699        .and_then(Value::as_u64)
700        .or_else(|| {
701            usage
702                .get("completion_tokens_details")
703                .and_then(|d| d.get("reasoning_tokens"))
704                .and_then(Value::as_u64)
705        });
706    // Only parsed counters establish usage; missing or malformed values are unknown.
707    if [
708        input_tokens,
709        output_tokens,
710        cache_read_tokens,
711        cache_write_tokens,
712        total_tokens,
713        reasoning_tokens,
714    ]
715    .iter()
716    .all(Option::is_none)
717    {
718        return None;
719    }
720    let input = input_tokens.unwrap_or_default();
721    let output = output_tokens.unwrap_or_default();
722    let cache_read = cache_read_tokens.unwrap_or_default();
723    let cache_write = cache_write_tokens.unwrap_or_default();
724    let total = total_tokens.unwrap_or_else(|| input.saturating_add(output));
725    Some(ParsedUsage {
726        usage: Usage {
727            input,
728            output,
729            cache_read,
730            cache_write,
731            total,
732            reasoning_tokens,
733        },
734        input_tokens,
735        reasoning_tokens,
736        presence: UsagePresence {
737            input: input_tokens.is_some(),
738            output: output_tokens.is_some(),
739            cache_read: cache_read_tokens.is_some(),
740            cache_write: cache_write_tokens.is_some(),
741            total: total_tokens.is_some(),
742            reasoning: reasoning_tokens.is_some(),
743        },
744    })
745}
magi_code/providers/stream.rs

magi_code/providers/
stream.rs
