codex_convert_proxy/convert/streaming/
converter.rs

1//! Core conversion logic: Chat SSE chunks to Responses API stream events.
2
3use crate::error::ConversionError;
4use crate::types::chat_api::{ChatStreamChunk, Content, ToolCallDelta};
5
6use super::events::ResponseStreamEvent;
7use super::state::{StreamState, ToolCallState};
8use super::super::util::{
9    map_tool_name_to_stream_item_type, parse_streaming_thinking, sanitize_pseudo_tool_markup,
10};
11
12/// Convert a Chat API SSE chunk to Responses API SSE events.
13pub fn chat_chunk_to_response_events(
14    chunk: &ChatStreamChunk,
15    state: &mut StreamState,
16) -> Result<Vec<ResponseStreamEvent>, ConversionError> {
17    let mut events = Vec::new();
18    let id = state.response_id.clone();
19    let model = chunk.model.as_deref().unwrap_or("unknown");
20
21    // On first chunk, emit created and in_progress
22    if state.is_first_chunk {
23        let created_at = chrono::Utc::now().timestamp();
24        events.push(ResponseStreamEvent::Created {
25            id: id.to_string(),
26            model: model.to_string(),
27            status: "in_progress".to_string(),
28            created_at,
29            request_context: state.request_context.clone(),
30        });
31        events.push(ResponseStreamEvent::InProgress {
32            id: id.to_string(),
33            model: model.to_string(),
34            status: "in_progress".to_string(),
35            created_at,
36            request_context: state.request_context.clone(),
37        });
38        state.is_first_chunk = false;
39    }
40
41    // Process each choice
42    for choice in &chunk.choices {
43        if let Some(delta) = &choice.delta {
44            tracing::debug!("[DELTA] content={:?}, tool_calls={:?}, function_call={:?}, refusal={:?}, reasoning_content={:?}",
45                delta.content.is_some(),
46                delta.tool_calls.as_ref().map(|tc| tc.len()),
47                delta.function_call.is_some(),
48                delta.refusal.as_ref().map(|r| r.len()),
49                delta.reasoning_content.as_ref().map(|r| r.len()));
50            // Handle reasoning content (GLM extension)
51            if let Some(reasoning) = &delta.reasoning_content
52                && !reasoning.is_empty() {
53                    if !state.is_reasoning_added {
54                        let reasoning_id = format!("reasoning_{}", id);
55                        let reasoning_idx = state.next_output_index;
56                        state.next_output_index += 1;
57                        state.reasoning_output_index = Some(reasoning_idx);
58                        events.push(ResponseStreamEvent::ReasoningAdded {
59                            output_index: reasoning_idx,
60                            item_id: reasoning_id.clone(),
61                        });
62                        state.is_reasoning_added = true;
63                    }
64                    let reasoning_idx = state.reasoning_output_index.unwrap_or(0);
65                    events.push(ResponseStreamEvent::ReasoningDelta {
66                        item_id: format!("reasoning_{}", id),
67                        output_index: reasoning_idx,
68                        content_index: 0,
69                        delta: reasoning.clone(),
70                    });
71                    state.reasoning_text.push_str(reasoning);
72                }
73
74            // Handle text content
75            if let Some(content) = &delta.content {
76                let text = match content {
77                    Content::String(s) => s.clone(),
78                    Content::Array(arr) => arr
79                        .iter()
80                        .filter_map(|b| b.text.clone())
81                        .collect::<Vec<_>>()
82                        .join(""),
83                };
84
85                if !text.is_empty() {
86                    // Parse thinking tags (<think> or <thought>) from content
87                    let (actual_text, reasoning_delta, new_is_thinking) =
88                        parse_streaming_thinking(&text, state.is_thinking, &mut state.thinking_buffer);
89                    let sanitized_actual_text = sanitize_pseudo_tool_markup(&actual_text);
90
91                    state.is_thinking = new_is_thinking;
92
93                    // Emit reasoning events if we have reasoning content
94                    if let Some(reasoning) = reasoning_delta
95                        && !reasoning.is_empty() {
96                            if !state.is_reasoning_added {
97                                let reasoning_id = format!("reasoning_{}", id);
98                                let reasoning_idx = state.next_output_index;
99                                state.next_output_index += 1;
100                                state.reasoning_output_index = Some(reasoning_idx);
101                                events.push(ResponseStreamEvent::ReasoningAdded {
102                                    output_index: reasoning_idx,
103                                    item_id: reasoning_id.clone(),
104                                });
105                                state.is_reasoning_added = true;
106                            }
107                            let reasoning_idx = state.reasoning_output_index.unwrap_or(0);
108                            events.push(ResponseStreamEvent::ReasoningDelta {
109                                item_id: format!("reasoning_{}", id),
110                                output_index: reasoning_idx,
111                                content_index: 0,
112                                delta: reasoning.clone(),
113                            });
114                            state.reasoning_text.push_str(&reasoning);
115                        }
116
117                    // Emit text events if we have actual content
118                    if !sanitized_actual_text.is_empty() {
119                        if !state.is_output_item_added {
120                            let text_idx = state.next_output_index;
121                            state.next_output_index += 1;
122                            state.text_output_index = Some(text_idx);
123                            events.push(ResponseStreamEvent::OutputItemAdded {
124                                output_index: text_idx,
125                                item_id: state.output_id.clone(),
126                                item_type: "message".to_string(),
127                                role: Some("assistant".to_string()),
128                                call_id: None,
129                                name: None,
130                            });
131                            events.push(ResponseStreamEvent::ContentPartAdded {
132                                item_id: state.output_id.clone(),
133                                output_index: text_idx,
134                                content_index: 0,
135                                part_type: "output_text".to_string(),
136                            });
137                            state.is_output_item_added = true;
138                            state.is_content_part_added = true;
139                        }
140
141                        let text_idx = state.text_output_index.unwrap_or(0);
142                        events.push(ResponseStreamEvent::OutputTextDelta {
143                            item_id: state.output_id.clone(),
144                            output_index: text_idx,
145                            content_index: 0,
146                            delta: sanitized_actual_text.clone(),
147                        });
148                        state.full_text.push_str(&sanitized_actual_text);
149                    }
150                }
151            }
152
153            // Handle refusal content
154            if let Some(refusal_delta) = &delta.refusal
155                && !refusal_delta.is_empty()
156            {
157                if !state.is_output_item_added {
158                    let text_idx = state.next_output_index;
159                    state.next_output_index += 1;
160                    state.text_output_index = Some(text_idx);
161                    events.push(ResponseStreamEvent::OutputItemAdded {
162                        output_index: text_idx,
163                        item_id: state.output_id.clone(),
164                        item_type: "message".to_string(),
165                        role: Some("assistant".to_string()),
166                        call_id: None,
167                        name: None,
168                    });
169                    events.push(ResponseStreamEvent::ContentPartAdded {
170                        item_id: state.output_id.clone(),
171                        output_index: text_idx,
172                        content_index: 0,
173                        part_type: "refusal".to_string(),
174                    });
175                    state.is_output_item_added = true;
176                    state.is_content_part_added = true;
177                }
178                let text_idx = state.text_output_index.unwrap_or(0);
179                events.push(ResponseStreamEvent::RefusalDelta {
180                    item_id: state.output_id.clone(),
181                    output_index: text_idx,
182                    content_index: 0,
183                    delta: refusal_delta.clone(),
184                });
185                state.refusal_text.push_str(refusal_delta);
186            }
187
188            // Handle tool calls
189            let mut normalized_tool_calls: Vec<ToolCallDelta> =
190                delta.tool_calls.clone().unwrap_or_default();
191            if normalized_tool_calls.is_empty()
192                && let Some(function_call) = delta.function_call.clone()
193            {
194                normalized_tool_calls.push(ToolCallDelta {
195                    index: 0,
196                    id: None,
197                    tool_type: Some("function".to_string()),
198                    function: function_call,
199                });
200            }
201            if !normalized_tool_calls.is_empty() {
202                tracing::debug!(
203                    "[TOOL_CALL] Processing {} tool calls in chunk",
204                    normalized_tool_calls.len()
205                );
206                for tc in &normalized_tool_calls {
207                    tracing::debug!("[TOOL_CALL] Tool call: id={:?}, index={}, name={:?}, args_len={}",
208                        tc.id, tc.index, tc.function.name, tc.function.arguments.as_ref().map(|a| a.len()).unwrap_or(0));
209
210                    let existing_idx = if let Some(tc_id) = tc.id.as_ref() {
211                        state.current_tool_calls.iter().position(|t| t.upstream_id.as_ref() == Some(tc_id))
212                    } else {
213                        state.current_tool_calls.iter().position(|t| t.chat_api_index == tc.index)
214                    };
215                    tracing::debug!("[TOOL_CALL] existing_idx={:?}, tc.index={}", existing_idx, tc.index);
216
217                    if existing_idx.is_none() {
218                        let tc_id = tc.id.clone().unwrap_or_else(|| {
219                            format!("call_{}_{}", tc.index, state.response_id)
220                        });
221                        let func_output_index = state.next_output_index;
222                        state.next_output_index += 1;
223                        let func_id = format!("func_{}_{}", func_output_index, state.response_id);
224                        let initial_name = tc.function.name.clone().unwrap_or_default();
225                        let item_type = map_tool_name_to_stream_item_type(&initial_name, state.request_context.as_ref());
226                        tracing::debug!("[TOOL_CALL] Creating new tool call: func_id={}, output_index={}", func_id, func_output_index);
227                        let name_for_item = if initial_name.is_empty() { None } else { Some(initial_name.clone()) };
228                        events.push(ResponseStreamEvent::OutputItemAdded {
229                            output_index: func_output_index,
230                            item_id: func_id.clone(),
231                            item_type: item_type.clone(),
232                            role: None,
233                            call_id: Some(tc_id.clone()),
234                            name: name_for_item,
235                        });
236                        state.is_function_call_item_added = true;
237
238                        let initial_args = tc.function.arguments.clone().unwrap_or_default();
239                        let tc_state = ToolCallState {
240                            upstream_id: tc.id.clone(),
241                            id: func_id.clone(),
242                            call_id: tc_id,
243                            item_type,
244                            name: initial_name,
245                            arguments: initial_args.clone(),
246                            output_index: func_output_index,
247                            chat_api_index: tc.index,
248                            last_args_len: initial_args.len(),
249                        };
250                        let call_id = tc_state.call_id.clone();
251
252                        state.current_tool_calls.push(tc_state);
253
254                        events.push(ResponseStreamEvent::FunctionCallArgumentsDelta {
255                            output_index: func_output_index,
256                            item_id: func_id,
257                            call_id: Some(call_id),
258                            delta: initial_args,
259                        });
260                        tracing::debug!("[TOOL_CALL] Emitted OutputItemAdded and FunctionCallArgumentsDelta, total events now: {}", events.len());
261                    } else if let Some(idx) = existing_idx {
262                        let tc_state = &mut state.current_tool_calls[idx];
263                        if let Some(args) = &tc.function.arguments {
264                            let prev_len = tc_state.last_args_len;
265                            let new_delta = if args.len() > prev_len && args.starts_with(&tc_state.arguments) {
266                                let delta = args[prev_len..].to_string();
267                                tc_state.arguments = args.clone();
268                                tc_state.last_args_len = args.len();
269                                delta
270                            } else {
271                                let delta = args.clone();
272                                tc_state.arguments.push_str(args);
273                                tc_state.last_args_len = tc_state.arguments.len();
274                                delta
275                            };
276
277                            if !new_delta.is_empty() {
278                                events.push(ResponseStreamEvent::FunctionCallArgumentsDelta {
279                                    output_index: tc_state.output_index,
280                                    item_id: tc_state.id.clone(),
281                                    call_id: Some(tc_state.call_id.clone()),
282                                    delta: new_delta,
283                                });
284                            }
285                        }
286                        if let Some(name) = &tc.function.name
287                            && !name.is_empty() && tc_state.name.is_empty() {
288                                tc_state.name = name.clone();
289                            }
290                    }
291                }
292            }
293
294            // Handle finish
295            tracing::debug!("[FINISH_REASON] choice.finish_reason={:?}, current_tool_calls_len={}", choice.finish_reason, state.current_tool_calls.len());
296            if let Some(reason) = &choice.finish_reason {
297                tracing::debug!("[FINISH_REASON] reason={}", reason);
298                if matches!(
299                    reason.as_str(),
300                    "stop" | "length" | "tool_calls" | "function_call" | "content_filter" | "refusal" | "refuse"
301                ) {
302                    apply_finish_reason(state, reason);
303                    events.extend(finalize_output(state, &id));
304                }
305            }
306        }
307    }
308
309    tracing::debug!("[CHUNK_EVENTS] Generated {} events: {:?}", events.len(),
310        events.iter().map(|e| format!("{:?}", e)).collect::<Vec<_>>());
311    Ok(events)
312}
313
314fn apply_finish_reason(state: &mut StreamState, reason: &str) {
315    match reason {
316        "length" => {
317            state.final_status = "incomplete".to_string();
318            state.incomplete_reason = Some("max_output_tokens".to_string());
319        }
320        "content_filter" => {
321            state.final_status = "incomplete".to_string();
322            state.incomplete_reason = Some("content_filter".to_string());
323        }
324        _ => {
325            state.final_status = "completed".to_string();
326            state.incomplete_reason = None;
327        }
328    }
329}
330
331/// Finalize output items when stream ends.
332fn finalize_output(state: &mut StreamState, id: &str) -> Vec<ResponseStreamEvent> {
333    let mut events = Vec::new();
334
335    tracing::debug!("[FINALIZE] is_output_item_added={}, is_reasoning_added={}, current_tool_calls={}",
336        state.is_output_item_added, state.is_reasoning_added, state.current_tool_calls.len());
337
338    // Finalize any pending tool calls
339    for tc_state in state.current_tool_calls.drain(..) {
340        events.push(ResponseStreamEvent::FunctionCallArgumentsDone {
341            output_index: tc_state.output_index,
342            item_id: tc_state.id.clone(),
343            call_id: tc_state.call_id.clone(),
344            arguments: tc_state.arguments.clone(),
345        });
346        events.push(ResponseStreamEvent::OutputItemDone {
347            output_index: tc_state.output_index,
348            item_id: tc_state.id.clone(),
349            item_type: tc_state.item_type.clone(),
350            role: None,
351            call_id: Some(tc_state.call_id.clone()),
352            name: Some(tc_state.name.clone()),
353            arguments: Some(tc_state.arguments.clone()),
354            text: None,
355            refusal: None,
356            summary: None,
357        });
358        state.completed_tool_calls.push(tc_state);
359    }
360
361    if state.is_output_item_added {
362        let text_idx = state.text_output_index.unwrap_or(0);
363        if !state.full_text.is_empty() {
364            events.push(ResponseStreamEvent::OutputTextDone {
365                item_id: state.output_id.clone(),
366                output_index: text_idx,
367                content_index: 0,
368                text: state.full_text.clone(),
369            });
370            events.push(ResponseStreamEvent::ContentPartDone {
371                item_id: state.output_id.clone(),
372                output_index: text_idx,
373                content_index: 0,
374                part_type: "output_text".to_string(),
375                text: state.full_text.clone(),
376            });
377        }
378        if !state.refusal_text.is_empty() {
379            events.push(ResponseStreamEvent::RefusalDone {
380                item_id: state.output_id.clone(),
381                output_index: text_idx,
382                content_index: 0,
383                refusal: state.refusal_text.clone(),
384            });
385            events.push(ResponseStreamEvent::ContentPartDone {
386                item_id: state.output_id.clone(),
387                output_index: text_idx,
388                content_index: 0,
389                part_type: "refusal".to_string(),
390                text: state.refusal_text.clone(),
391            });
392        }
393        events.push(ResponseStreamEvent::OutputItemDone {
394            output_index: text_idx,
395            item_id: state.output_id.clone(),
396            item_type: "message".to_string(),
397            role: Some("assistant".to_string()),
398            call_id: None,
399            name: None,
400            arguments: None,
401            text: if state.full_text.is_empty() {
402                None
403            } else {
404                Some(state.full_text.clone())
405            },
406            refusal: if state.refusal_text.is_empty() {
407                None
408            } else {
409                Some(state.refusal_text.clone())
410            },
411            summary: None,
412        });
413    }
414
415    if state.is_reasoning_added {
416        let reasoning_idx = state.reasoning_output_index.unwrap_or(0);
417        let reasoning_id = format!("reasoning_{}", id);
418        events.push(ResponseStreamEvent::ReasoningTextDone {
419            item_id: reasoning_id.clone(),
420            output_index: reasoning_idx,
421            content_index: 0,
422            text: state.reasoning_text.clone(),
423        });
424        events.push(ResponseStreamEvent::OutputItemDone {
425            output_index: reasoning_idx,
426            item_id: reasoning_id,
427            item_type: "reasoning".to_string(),
428            role: None,
429            call_id: None,
430            name: None,
431            arguments: None,
432            text: None,
433            refusal: None,
434            summary: Some(vec![crate::types::response_api::ReasoningSummaryPart::SummaryText {
435                text: state.reasoning_text.clone(),
436            }]),
437        });
438    }
439
440    tracing::debug!("[FINALIZE] Produced {} events", events.len());
441    events
442}
443
444#[cfg(test)]
445mod tests {
446    use super::*;
447    use crate::types::chat_api::{ChatDelta, ChatStreamChoice, Content, ToolCallDelta, FunctionCallDelta};
448
449    #[test]
450    fn test_first_chunk_generates_created_event() {
451        let chunk = ChatStreamChunk {
452            id: Some("chat_123".to_string()),
453            object: Some("chat.completion.chunk".to_string()),
454            created: Some(1234567890),
455            model: Some("gpt-4o".to_string()),
456            choices: vec![ChatStreamChoice {
457                index: 0,
458                delta: Some(ChatDelta {
459                    role: Some("assistant".to_string()),
460                    content: Some(Content::String("Hello".to_string())),
461                    tool_calls: None,
462                    function_call: None,
463                    reasoning_content: None,
464                    refusal: None,
465                }),
466                finish_reason: None,
467            }],
468            usage: None,
469        };
470
471        let mut state = StreamState::new("chat_123".to_string(), "gpt-4o".to_string(), None);
472        let events = chat_chunk_to_response_events(&chunk, &mut state).unwrap();
473
474        assert!(events.iter().any(|e| matches!(e, ResponseStreamEvent::Created { .. })));
475        assert!(events.iter().any(|e| matches!(e, ResponseStreamEvent::InProgress { .. })));
476        assert!(events.iter().any(|e| matches!(e, ResponseStreamEvent::OutputTextDelta { delta, .. } if delta == "Hello")));
477    }
478
479    #[test]
480    fn test_tool_call_generates_function_call_events() {
481        let chunk = ChatStreamChunk {
482            id: Some("chat_123".to_string()),
483            object: Some("chat.completion.chunk".to_string()),
484            created: Some(1234567890),
485            model: Some("gpt-4o".to_string()),
486            choices: vec![ChatStreamChoice {
487                index: 0,
488                delta: Some(ChatDelta {
489                    role: Some("assistant".to_string()),
490                    content: None,
491                    tool_calls: Some(vec![ToolCallDelta {
492                        index: 0,
493                        id: Some("call_abc".to_string()),
494                        tool_type: Some("function".to_string()),
495                        function: FunctionCallDelta {
496                            name: Some("get_weather".to_string()),
497                            arguments: Some(r#"{"city":"Beijing"}"#.to_string()),
498                        },
499                    }]),
500                    function_call: None,
501                    reasoning_content: None,
502                    refusal: None,
503                }),
504                finish_reason: None,
505            }],
506            usage: None,
507        };
508
509        let mut state = StreamState::new("chat_123".to_string(), "gpt-4o".to_string(), None);
510        let _ = chat_chunk_to_response_events(&chunk, &mut state);
511
512        assert!(!state.current_tool_calls.is_empty());
513        let tc = state.current_tool_calls.first().unwrap();
514        assert_eq!(tc.name, "get_weather");
515    }
516
517    #[test]
518    fn test_parse_streaming_thinking_basic() {
519        use crate::convert::util::parse_streaming_thinking;
520        let mut buffer = String::new();
521        let (actual, reasoning, is_thinking) = parse_streaming_thinking("Hello world", false, &mut buffer);
522        assert_eq!(actual, "Hello world");
523        assert!(reasoning.is_none());
524        assert!(!is_thinking);
525    }
526
527    #[test]
528    fn test_parse_streaming_thinking_with_think_tag() {
529        use crate::convert::util::parse_streaming_thinking;
530        let mut buffer = String::new();
531        let (actual, reasoning, is_thinking) = parse_streaming_thinking(
532            "<think>\nreasoning\n</think>\n\nactual text",
533            false,
534            &mut buffer,
535        );
536        assert_eq!(actual, "\n\nactual text");
537        assert_eq!(reasoning, Some("\nreasoning\n".to_string()));
538        assert!(!is_thinking);
539    }
540
541    #[test]
542    fn test_parse_streaming_thinking_chunked() {
543        use crate::convert::util::parse_streaming_thinking;
544        let mut buffer = String::new();
545
546        let (actual, reasoning, is_thinking) = parse_streaming_thinking(
547            "<think>\npartial",
548            false,
549            &mut buffer,
550        );
551        assert_eq!(actual, "");
552        assert!(reasoning.is_none());
553        assert!(is_thinking);
554
555        let (actual, reasoning, is_thinking) = parse_streaming_thinking(
556            " content\n</think>\n\nfinal",
557            is_thinking,
558            &mut buffer,
559        );
560        assert_eq!(actual, "\n\nfinal");
561        assert_eq!(reasoning, Some("\npartial content\n".to_string()));
562        assert!(!is_thinking);
563    }
564
565    #[test]
566    fn test_parse_streaming_thought_tag() {
567        use crate::convert::util::parse_streaming_thinking;
568        let mut buffer = String::new();
569        let (actual, reasoning, is_thinking) = parse_streaming_thinking(
570            "<thought>reasoning</thought>actual",
571            false,
572            &mut buffer,
573        );
574        assert_eq!(actual, "actual");
575        assert_eq!(reasoning, Some("reasoning".to_string()));
576        assert!(!is_thinking);
577    }
578
579    #[test]
580    fn test_finish_reason_content_filter_marks_incomplete() {
581        let mut state = StreamState::new("chat_123".to_string(), "gpt-4o".to_string(), None);
582        let chunk = ChatStreamChunk {
583            id: Some("chat_123".to_string()),
584            object: Some("chat.completion.chunk".to_string()),
585            created: Some(1234567890),
586            model: Some("gpt-4o".to_string()),
587            choices: vec![ChatStreamChoice {
588                index: 0,
589                delta: Some(ChatDelta {
590                    role: Some("assistant".to_string()),
591                    content: Some(Content::String("partial".to_string())),
592                    tool_calls: None,
593                    function_call: None,
594                    reasoning_content: None,
595                    refusal: None,
596                }),
597                finish_reason: Some("content_filter".to_string()),
598            }],
599            usage: None,
600        };
601        let _ = chat_chunk_to_response_events(&chunk, &mut state).unwrap();
602        assert_eq!(state.final_status, "incomplete");
603        assert_eq!(state.incomplete_reason.as_deref(), Some("content_filter"));
604    }
605
606    #[test]
607    fn test_refusal_delta_emits_refusal_event() {
608        let mut state = StreamState::new("chat_123".to_string(), "gpt-4o".to_string(), None);
609        let chunk = ChatStreamChunk {
610            id: Some("chat_123".to_string()),
611            object: Some("chat.completion.chunk".to_string()),
612            created: Some(1234567890),
613            model: Some("gpt-4o".to_string()),
614            choices: vec![ChatStreamChoice {
615                index: 0,
616                delta: Some(ChatDelta {
617                    role: Some("assistant".to_string()),
618                    content: None,
619                    tool_calls: None,
620                    function_call: None,
621                    reasoning_content: None,
622                    refusal: Some("I cannot comply".to_string()),
623                }),
624                finish_reason: Some("refusal".to_string()),
625            }],
626            usage: None,
627        };
628        let events = chat_chunk_to_response_events(&chunk, &mut state).unwrap();
629        assert!(events
630            .iter()
631            .any(|e| matches!(e, ResponseStreamEvent::RefusalDelta { .. })));
632    }
633
634    #[test]
635    fn test_legacy_function_call_delta_supported() {
636        let mut state = StreamState::new("chat_123".to_string(), "gpt-4o".to_string(), None);
637        let chunk = ChatStreamChunk {
638            id: Some("chat_123".to_string()),
639            object: Some("chat.completion.chunk".to_string()),
640            created: Some(1234567890),
641            model: Some("gpt-4o".to_string()),
642            choices: vec![ChatStreamChoice {
643                index: 0,
644                delta: Some(ChatDelta {
645                    role: Some("assistant".to_string()),
646                    content: None,
647                    tool_calls: None,
648                    function_call: Some(FunctionCallDelta {
649                        name: Some("get_weather".to_string()),
650                        arguments: Some(r#"{"city":"Beijing"}"#.to_string()),
651                    }),
652                    reasoning_content: None,
653                    refusal: None,
654                }),
655                finish_reason: Some("function_call".to_string()),
656            }],
657            usage: None,
658        };
659        let _ = chat_chunk_to_response_events(&chunk, &mut state).unwrap();
660        assert!(!state.completed_tool_calls.is_empty());
661    }
662}
codex_convert_proxy/convert/streaming/converter.rs

codex_convert_proxy/convert/streaming/
converter.rs
