a3s_code_core/llm/
openai.rs

1//! OpenAI-compatible LLM client
2
3use super::http::{default_http_client, normalize_base_url, HttpClient};
4use super::structured;
5use super::types::*;
6use super::LlmClient;
7use crate::llm::types::{ToolResultContent, ToolResultContentField};
8use crate::retry::{AttemptOutcome, RetryConfig};
9use anyhow::{Context, Result};
10use async_trait::async_trait;
11use futures::StreamExt;
12use serde::Deserialize;
13use std::collections::HashMap;
14use std::sync::Arc;
15use std::time::Instant;
16use tokio::sync::mpsc;
17
18/// OpenAI client
19pub struct OpenAiClient {
20    pub(crate) provider_name: String,
21    pub(crate) api_key: SecretString,
22    pub(crate) model: String,
23    pub(crate) base_url: String,
24    pub(crate) chat_completions_path: String,
25    pub(crate) headers: HashMap<String, String>,
26    pub(crate) temperature: Option<f32>,
27    pub(crate) max_tokens: Option<usize>,
28    pub(crate) logprobs: bool,
29    pub(crate) top_logprobs: Option<usize>,
30    pub(crate) http: Arc<dyn HttpClient>,
31    pub(crate) retry_config: RetryConfig,
32    pub(crate) native_structured_support: structured::NativeStructuredSupport,
33}
34
35impl OpenAiClient {
36    pub(crate) fn parse_tool_arguments(tool_name: &str, arguments: &str) -> serde_json::Value {
37        if arguments.trim().is_empty() {
38            return serde_json::Value::Object(Default::default());
39        }
40
41        serde_json::from_str(arguments).unwrap_or_else(|e| {
42            tracing::warn!(
43                "Failed to parse tool arguments JSON for tool '{}': {}",
44                tool_name,
45                e
46            );
47            serde_json::json!({
48                "__parse_error": format!(
49                    "Malformed tool arguments: {}. Raw input: {}",
50                    e, arguments
51                )
52            })
53        })
54    }
55
56    fn merge_stream_text(text_content: &mut String, incoming: &str) -> Option<String> {
57        if incoming.is_empty() {
58            return None;
59        }
60        if text_content.is_empty() {
61            text_content.push_str(incoming);
62            return Some(incoming.to_string());
63        }
64        if incoming == text_content.as_str() || text_content.ends_with(incoming) {
65            return None;
66        }
67        // If incoming contains text_content as a prefix (incoming is the full content),
68        // replace text_content instead of appending (avoids duplicate full content)
69        if incoming.starts_with(text_content.as_str()) && incoming.len() > text_content.len() {
70            let suffix = &incoming[text_content.len()..];
71            if !suffix.is_empty() {
72                *text_content = incoming.to_string();
73                return Some(suffix.to_string());
74            }
75            return None;
76        }
77        if let Some(suffix) = incoming.strip_prefix(text_content.as_str()) {
78            if suffix.is_empty() {
79                return None;
80            }
81            text_content.push_str(suffix);
82            return Some(suffix.to_string());
83        }
84        text_content.push_str(incoming);
85        Some(incoming.to_string())
86    }
87
88    pub fn new(api_key: String, model: String) -> Self {
89        Self {
90            provider_name: "openai".to_string(),
91            api_key: SecretString::new(api_key),
92            model,
93            base_url: "https://api.openai.com".to_string(),
94            chat_completions_path: "/v1/chat/completions".to_string(),
95            headers: HashMap::new(),
96            temperature: None,
97            max_tokens: None,
98            logprobs: false,
99            top_logprobs: None,
100            http: default_http_client(),
101            retry_config: RetryConfig::default(),
102            native_structured_support: structured::NativeStructuredSupport::JsonSchema,
103        }
104    }
105
106    pub fn with_base_url(mut self, base_url: String) -> Self {
107        self.base_url = normalize_base_url(&base_url);
108        self
109    }
110
111    pub fn with_provider_name(mut self, provider_name: impl Into<String>) -> Self {
112        self.provider_name = provider_name.into();
113        self
114    }
115
116    pub fn with_chat_completions_path(mut self, path: impl Into<String>) -> Self {
117        let path = path.into();
118        self.chat_completions_path = if path.starts_with('/') {
119            path
120        } else {
121            format!("/{}", path)
122        };
123        self
124    }
125
126    pub fn with_temperature(mut self, temperature: f32) -> Self {
127        self.temperature = Some(temperature);
128        self
129    }
130
131    pub fn with_headers(mut self, headers: HashMap<String, String>) -> Self {
132        self.headers = headers;
133        self
134    }
135
136    pub fn with_max_tokens(mut self, max_tokens: usize) -> Self {
137        self.max_tokens = Some(max_tokens);
138        self
139    }
140
141    pub fn with_logprobs(mut self, enabled: bool) -> Self {
142        self.logprobs = enabled;
143        self
144    }
145
146    pub fn with_top_logprobs(mut self, top_logprobs: usize) -> Self {
147        self.logprobs = true;
148        self.top_logprobs = Some(top_logprobs);
149        self
150    }
151
152    pub fn with_retry_config(mut self, retry_config: RetryConfig) -> Self {
153        self.retry_config = retry_config;
154        self
155    }
156
157    pub fn with_native_structured_support(
158        mut self,
159        support: structured::NativeStructuredSupport,
160    ) -> Self {
161        self.native_structured_support = support;
162        self
163    }
164
165    pub fn with_http_client(mut self, http: Arc<dyn HttpClient>) -> Self {
166        self.http = http;
167        self
168    }
169
170    pub(crate) fn request_headers(&self) -> Vec<(String, String)> {
171        let mut headers = Vec::with_capacity(self.headers.len() + 1);
172        let has_authorization = self
173            .headers
174            .keys()
175            .any(|key| key.eq_ignore_ascii_case("authorization"));
176        if !has_authorization {
177            headers.push((
178                "Authorization".to_string(),
179                format!("Bearer {}", self.api_key.expose()),
180            ));
181        }
182        headers.extend(
183            self.headers
184                .iter()
185                .map(|(key, value)| (key.clone(), value.clone())),
186        );
187        headers
188    }
189
190    pub(crate) fn convert_messages(&self, messages: &[Message]) -> Vec<serde_json::Value> {
191        messages
192            .iter()
193            .map(|msg| {
194                let content: serde_json::Value = if msg.content.len() == 1 {
195                    match &msg.content[0] {
196                        ContentBlock::Text { text } => serde_json::json!(text),
197                        ContentBlock::ToolResult {
198                            tool_use_id,
199                            content,
200                            ..
201                        } => {
202                            let content_str = match content {
203                                ToolResultContentField::Text(s) => s.clone(),
204                                ToolResultContentField::Blocks(blocks) => blocks
205                                    .iter()
206                                    .filter_map(|b| {
207                                        if let ToolResultContent::Text { text } = b {
208                                            Some(text.clone())
209                                        } else {
210                                            None
211                                        }
212                                    })
213                                    .collect::<Vec<_>>()
214                                    .join("\n"),
215                            };
216                            return serde_json::json!({
217                                "role": "tool",
218                                "tool_call_id": tool_use_id,
219                                "content": content_str,
220                            });
221                        }
222                        _ => serde_json::json!(""),
223                    }
224                } else {
225                    serde_json::json!(msg
226                        .content
227                        .iter()
228                        .map(|block| {
229                            match block {
230                                ContentBlock::Text { text } => serde_json::json!({
231                                    "type": "text",
232                                    "text": text,
233                                }),
234                                ContentBlock::Image { source } => serde_json::json!({
235                                    "type": "image_url",
236                                    "image_url": {
237                                        "url": format!(
238                                            "data:{};base64,{}",
239                                            source.media_type, source.data
240                                        ),
241                                    }
242                                }),
243                                ContentBlock::ToolUse { id, name, input } => serde_json::json!({
244                                    "type": "function",
245                                    "id": id,
246                                    "function": {
247                                        "name": name,
248                                        "arguments": input.to_string(),
249                                    }
250                                }),
251                                _ => serde_json::json!({}),
252                            }
253                        })
254                        .collect::<Vec<_>>())
255                };
256
257                // Handle assistant messages — kimi-k2.5 requires reasoning_content
258                // on all assistant messages when thinking mode is enabled
259                if msg.role == "assistant" {
260                    let rc = msg.reasoning_content.as_deref().unwrap_or("");
261                    let tool_calls: Vec<_> = msg.tool_calls();
262                    if !tool_calls.is_empty() {
263                        return serde_json::json!({
264                            "role": "assistant",
265                            "content": msg.text(),
266                            "reasoning_content": rc,
267                            "tool_calls": tool_calls.iter().map(|tc| {
268                                serde_json::json!({
269                                    "id": tc.id,
270                                    "type": "function",
271                                    "function": {
272                                        "name": tc.name,
273                                        "arguments": tc.args.to_string(),
274                                    }
275                                })
276                            }).collect::<Vec<_>>(),
277                        });
278                    }
279                    return serde_json::json!({
280                        "role": "assistant",
281                        "content": content,
282                        "reasoning_content": rc,
283                    });
284                }
285
286                serde_json::json!({
287                    "role": msg.role,
288                    "content": content,
289                })
290            })
291            .collect()
292    }
293
294    pub(crate) fn convert_tools(&self, tools: &[ToolDefinition]) -> Vec<serde_json::Value> {
295        tools
296            .iter()
297            .map(|t| {
298                serde_json::json!({
299                    "type": "function",
300                    "function": {
301                        "name": t.name,
302                        "description": t.description,
303                        "parameters": t.parameters,
304                    }
305                })
306            })
307            .collect()
308    }
309}
310
311impl OpenAiClient {
312    /// Apply a structured-output directive to an OpenAI-compatible request.
313    ///
314    /// OpenAI-compatible APIs support both forced function `tool_choice` and
315    /// native `response_format` (`json_object` / `json_schema` + `strict`).
316    fn apply_directive(
317        request: &mut serde_json::Value,
318        directive: &structured::StructuredDirective,
319    ) {
320        if let Some(tool) = &directive.force_tool {
321            request["tool_choice"] = serde_json::json!({
322                "type": "function",
323                "function": { "name": tool }
324            });
325        }
326        if let Some(rf) = &directive.response_format {
327            request["response_format"] = match rf {
328                structured::ResponseFormat::JsonObject => {
329                    serde_json::json!({ "type": "json_object" })
330                }
331                structured::ResponseFormat::JsonSchema { name, schema } => serde_json::json!({
332                    "type": "json_schema",
333                    "json_schema": { "name": name, "schema": schema, "strict": true }
334                }),
335            };
336        }
337    }
338
339    /// Build a chat-completions request body, optionally applying a directive.
340    fn build_chat_request(
341        &self,
342        messages: &[Message],
343        system: Option<&str>,
344        tools: &[ToolDefinition],
345        directive: Option<&structured::StructuredDirective>,
346    ) -> serde_json::Value {
347        let mut openai_messages = Vec::new();
348
349        if let Some(sys) = system {
350            openai_messages.push(serde_json::json!({
351                "role": "system",
352                "content": sys,
353            }));
354        }
355
356        openai_messages.extend(self.convert_messages(messages));
357
358        let mut request = serde_json::json!({
359            "model": self.model,
360            "messages": openai_messages,
361        });
362
363        if let Some(temp) = self.temperature {
364            request["temperature"] = serde_json::json!(temp);
365        }
366        if let Some(max) = self.max_tokens {
367            request["max_tokens"] = serde_json::json!(max);
368        }
369        if self.logprobs {
370            request["logprobs"] = serde_json::json!(true);
371            if let Some(top_logprobs) = self.top_logprobs {
372                request["top_logprobs"] = serde_json::json!(top_logprobs);
373            }
374        }
375
376        if !tools.is_empty() {
377            request["tools"] = serde_json::json!(self.convert_tools(tools));
378        }
379
380        if let Some(directive) = directive {
381            Self::apply_directive(&mut request, directive);
382        }
383
384        request
385    }
386
387    /// Execute a fully-built (non-streaming) chat-completions request.
388    async fn send_request(&self, request: serde_json::Value) -> Result<LlmResponse> {
389        {
390            let request_started_at = Instant::now();
391            let url = format!("{}{}", self.base_url, self.chat_completions_path);
392            let request_headers = self.request_headers();
393
394            let response = crate::retry::with_retry(&self.retry_config, |_attempt| {
395                let http = &self.http;
396                let url = &url;
397                let request_headers = request_headers.clone();
398                let request = &request;
399                async move {
400                    let headers = request_headers
401                        .iter()
402                        .map(|(key, value)| (key.as_str(), value.as_str()))
403                        .collect::<Vec<_>>();
404                    // Non-streaming: use a non-cancelled token for now
405                    let cancel_token = tokio_util::sync::CancellationToken::new();
406                    match http.post(url, headers, request, cancel_token).await {
407                        Ok(resp) => {
408                            let status = reqwest::StatusCode::from_u16(resp.status)
409                                .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
410                            if status.is_success() {
411                                AttemptOutcome::Success(resp.body)
412                            } else if self.retry_config.is_retryable_status(status) {
413                                AttemptOutcome::Retryable {
414                                    status,
415                                    body: resp.body,
416                                    retry_after: None,
417                                }
418                            } else {
419                                AttemptOutcome::Fatal(anyhow::anyhow!(
420                                    "OpenAI API error at {} ({}): {}",
421                                    url,
422                                    status,
423                                    resp.body
424                                ))
425                            }
426                        }
427                        Err(e) => {
428                            tracing::error!("HTTP error: {e:?}");
429                            AttemptOutcome::Fatal(e)
430                        }
431                    }
432                }
433            })
434            .await?;
435
436            let parsed: OpenAiResponse =
437                serde_json::from_str(&response).context("Failed to parse OpenAI response")?;
438
439            let choice = parsed.choices.into_iter().next().context("No choices")?;
440            let token_logprobs = choice
441                .logprobs
442                .as_ref()
443                .map(openai_logprobs_to_token_logprobs)
444                .unwrap_or_default();
445
446            let mut content = vec![];
447
448            let reasoning_content = choice.message.reasoning_content;
449
450            let text_content = choice.message.content;
451
452            if let Some(text) = text_content {
453                if !text.is_empty() {
454                    content.push(ContentBlock::Text { text });
455                }
456            }
457
458            if let Some(tool_calls) = choice.message.tool_calls {
459                for tc in tool_calls {
460                    content.push(ContentBlock::ToolUse {
461                        id: tc.id,
462                        name: tc.function.name.clone(),
463                        input: Self::parse_tool_arguments(
464                            &tc.function.name,
465                            &tc.function.arguments,
466                        ),
467                    });
468                }
469            }
470
471            let llm_response = LlmResponse {
472                message: Message {
473                    role: "assistant".to_string(),
474                    content,
475                    reasoning_content,
476                },
477                usage: TokenUsage {
478                    prompt_tokens: parsed.usage.prompt_tokens,
479                    completion_tokens: parsed.usage.completion_tokens,
480                    total_tokens: {
481                        let t = parsed.usage.total_tokens;
482                        // MiniMax: fall back to total_characters when total_tokens is 0.
483                        if t == 0 {
484                            parsed.usage.total_characters.unwrap_or(0)
485                        } else {
486                            t
487                        }
488                    },
489                    cache_read_tokens: parsed
490                        .usage
491                        .prompt_tokens_details
492                        .as_ref()
493                        .and_then(|d| d.cached_tokens),
494                    cache_write_tokens: None,
495                },
496                stop_reason: choice.finish_reason,
497                token_logprobs,
498                meta: Some(LlmResponseMeta {
499                    provider: Some(self.provider_name.clone()),
500                    request_model: Some(self.model.clone()),
501                    request_url: Some(url.clone()),
502                    response_id: parsed.id,
503                    response_model: parsed.model,
504                    response_object: parsed.object,
505                    first_token_ms: None,
506                    duration_ms: Some(request_started_at.elapsed().as_millis() as u64),
507                }),
508            };
509
510            crate::telemetry::record_llm_usage(
511                llm_response.usage.prompt_tokens,
512                llm_response.usage.completion_tokens,
513                llm_response.usage.total_tokens,
514                llm_response.stop_reason.as_deref(),
515            );
516
517            Ok(llm_response)
518        }
519    }
520}
521
522#[async_trait]
523impl LlmClient for OpenAiClient {
524    async fn complete(
525        &self,
526        messages: &[Message],
527        system: Option<&str>,
528        tools: &[ToolDefinition],
529    ) -> Result<LlmResponse> {
530        self.send_request(self.build_chat_request(messages, system, tools, None))
531            .await
532    }
533
534    async fn complete_structured(
535        &self,
536        messages: &[Message],
537        system: Option<&str>,
538        tools: &[ToolDefinition],
539        directive: &structured::StructuredDirective,
540    ) -> Result<LlmResponse> {
541        self.send_request(self.build_chat_request(messages, system, tools, Some(directive)))
542            .await
543    }
544
545    fn native_structured_support(&self) -> structured::NativeStructuredSupport {
546        self.native_structured_support
547    }
548
549    async fn complete_streaming(
550        &self,
551        messages: &[Message],
552        system: Option<&str>,
553        tools: &[ToolDefinition],
554        cancel_token: tokio_util::sync::CancellationToken,
555    ) -> Result<mpsc::Receiver<StreamEvent>> {
556        self.send_streaming(
557            self.build_chat_request(messages, system, tools, None),
558            cancel_token,
559        )
560        .await
561    }
562
563    async fn complete_streaming_structured(
564        &self,
565        messages: &[Message],
566        system: Option<&str>,
567        tools: &[ToolDefinition],
568        directive: &structured::StructuredDirective,
569        cancel_token: tokio_util::sync::CancellationToken,
570    ) -> Result<mpsc::Receiver<StreamEvent>> {
571        self.send_streaming(
572            self.build_chat_request(messages, system, tools, Some(directive)),
573            cancel_token,
574        )
575        .await
576    }
577}
578
579impl OpenAiClient {
580    /// Execute a fully-built streaming chat-completions request (sets `stream`).
581    async fn send_streaming(
582        &self,
583        mut request: serde_json::Value,
584        cancel_token: tokio_util::sync::CancellationToken,
585    ) -> Result<mpsc::Receiver<StreamEvent>> {
586        {
587            request["stream"] = serde_json::json!(true);
588            request["stream_options"] = serde_json::json!({ "include_usage": true });
589            let request_started_at = Instant::now();
590            let url = format!("{}{}", self.base_url, self.chat_completions_path);
591            let request_headers = self.request_headers();
592
593            let streaming_resp = crate::retry::with_retry(&self.retry_config, |_attempt| {
594                let http = &self.http;
595                let url = &url;
596                let request_headers = request_headers.clone();
597                let request = &request;
598                let cancel_token = cancel_token.clone();
599                async move {
600                    let headers = request_headers
601                        .iter()
602                        .map(|(key, value)| (key.as_str(), value.as_str()))
603                        .collect::<Vec<_>>();
604                    // Wrap in tokio::select! so cancellation aborts the HTTP request mid-flight
605                    let resp = tokio::select! {
606                        _ = cancel_token.cancelled() => {
607                            return AttemptOutcome::Fatal(anyhow::anyhow!("HTTP request cancelled"));
608                        }
609                        result = http.post_streaming(url, headers, request, cancel_token.clone()) => {
610                            match result {
611                                Ok(r) => r,
612                                Err(e) => {
613                                    // Transient network error (timeout, reset,
614                                    // mid-flight drop — common on throttled
615                                    // endpoints): retry with backoff like 429/5xx
616                                    // instead of failing the turn. GLM and other
617                                    // OpenAI-compatible endpoints hit this most.
618                                    return if crate::retry::is_transient_error(&e) {
619                                        AttemptOutcome::Retryable {
620                                            status: reqwest::StatusCode::SERVICE_UNAVAILABLE,
621                                            body: format!("network error: {e}"),
622                                            retry_after: None,
623                                        }
624                                    } else {
625                                        AttemptOutcome::Fatal(anyhow::anyhow!(
626                                            "HTTP request failed: {}",
627                                            e
628                                        ))
629                                    };
630                                }
631                            }
632                        }
633                    };
634                    let status = reqwest::StatusCode::from_u16(resp.status)
635                        .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR);
636                    if status.is_success() {
637                        AttemptOutcome::Success(resp)
638                    } else {
639                        let retry_after = resp
640                            .retry_after
641                            .as_deref()
642                            .and_then(|v| RetryConfig::parse_retry_after(Some(v)));
643                        if self.retry_config.is_retryable_status(status) {
644                            AttemptOutcome::Retryable {
645                                status,
646                                body: resp.error_body,
647                                retry_after,
648                            }
649                        } else {
650                            AttemptOutcome::Fatal(anyhow::anyhow!(
651                                "OpenAI API error at {} ({}): {}",
652                                url,
653                                status,
654                                resp.error_body
655                            ))
656                        }
657                    }
658                }
659            })
660            .await?;
661
662            let (tx, rx) = mpsc::channel(100);
663
664            let mut stream = streaming_resp.byte_stream;
665            let provider_name = self.provider_name.clone();
666            let request_model = self.model.clone();
667            let request_url = url.clone();
668            tokio::spawn(async move {
669                let mut buffer = String::new();
670                let mut content_blocks: Vec<ContentBlock> = Vec::new();
671                let mut text_content = String::new();
672                let mut reasoning_content_accum = String::new();
673                let mut tool_calls: std::collections::BTreeMap<usize, (String, String, String)> =
674                    std::collections::BTreeMap::new();
675                let mut usage = TokenUsage::default();
676                let mut finish_reason = None;
677                let mut token_logprobs: Vec<TokenLogProb> = Vec::new();
678                let mut response_id = None;
679                let mut response_model = None;
680                let mut response_object = None;
681                let mut first_token_ms = None;
682                let mut saw_done = false;
683                let mut parsed_any_event = false;
684
685                while let Some(chunk_result) = stream.next().await {
686                    let chunk = match chunk_result {
687                        Ok(c) => c,
688                        Err(e) => {
689                            tracing::error!("Stream error: {}", e);
690                            break;
691                        }
692                    };
693
694                    buffer.push_str(&String::from_utf8_lossy(&chunk));
695
696                    while let Some(event_end) = buffer.find("\n\n") {
697                        let event_data: String = buffer.drain(..event_end).collect();
698                        buffer.drain(..2);
699
700                        for line in event_data.lines() {
701                            if let Some(data) = crate::sse::data_field_value(line) {
702                                if data == "[DONE]" {
703                                    saw_done = true;
704                                    if !text_content.is_empty() {
705                                        content_blocks.push(ContentBlock::Text {
706                                            text: text_content.clone(),
707                                        });
708                                    }
709                                    for (_, (id, name, args)) in tool_calls.iter() {
710                                        content_blocks.push(ContentBlock::ToolUse {
711                                            id: id.clone(),
712                                            name: name.clone(),
713                                            input: Self::parse_tool_arguments(name, args),
714                                        });
715                                    }
716                                    tool_calls.clear();
717                                    crate::telemetry::record_llm_usage(
718                                        usage.prompt_tokens,
719                                        usage.completion_tokens,
720                                        usage.total_tokens,
721                                        finish_reason.as_deref(),
722                                    );
723                                    let response = LlmResponse {
724                                        message: Message {
725                                            role: "assistant".to_string(),
726                                            content: std::mem::take(&mut content_blocks),
727                                            reasoning_content: if reasoning_content_accum.is_empty()
728                                            {
729                                                None
730                                            } else {
731                                                Some(std::mem::take(&mut reasoning_content_accum))
732                                            },
733                                        },
734                                        usage: usage.clone(),
735                                        stop_reason: std::mem::take(&mut finish_reason),
736                                        token_logprobs: std::mem::take(&mut token_logprobs),
737                                        meta: Some(LlmResponseMeta {
738                                            provider: Some(provider_name.clone()),
739                                            request_model: Some(request_model.clone()),
740                                            request_url: Some(request_url.clone()),
741                                            response_id: response_id.clone(),
742                                            response_model: response_model.clone(),
743                                            response_object: response_object.clone(),
744                                            first_token_ms,
745                                            duration_ms: Some(
746                                                request_started_at.elapsed().as_millis() as u64,
747                                            ),
748                                        }),
749                                    };
750                                    let _ = tx.send(StreamEvent::Done(response)).await;
751                                    continue;
752                                }
753
754                                if let Ok(event) = serde_json::from_str::<OpenAiStreamChunk>(data) {
755                                    parsed_any_event = true;
756                                    if response_id.is_none() {
757                                        response_id = event.id.clone();
758                                    }
759                                    if response_model.is_none() {
760                                        response_model = event.model.clone();
761                                    }
762                                    if response_object.is_none() {
763                                        response_object = event.object.clone();
764                                    }
765                                    if let Some(u) = event.usage {
766                                        usage.prompt_tokens = u.prompt_tokens;
767                                        usage.completion_tokens = u.completion_tokens;
768                                        usage.total_tokens = u.total_tokens;
769                                        // MiniMax: fall back to total_characters when total_tokens is 0.
770                                        if usage.total_tokens == 0 {
771                                            usage.total_tokens = u.total_characters.unwrap_or(0);
772                                        }
773                                        usage.cache_read_tokens = u
774                                            .prompt_tokens_details
775                                            .as_ref()
776                                            .and_then(|d| d.cached_tokens);
777                                    }
778
779                                    if let Some(choice) = event.choices.into_iter().next() {
780                                        if let Some(logprobs) = choice.logprobs.as_ref() {
781                                            token_logprobs.extend(
782                                                openai_logprobs_to_token_logprobs(logprobs),
783                                            );
784                                        }
785                                        if let Some(reason) = choice.finish_reason {
786                                            finish_reason = Some(reason);
787                                        }
788
789                                        if let Some(message) = choice.message {
790                                            // If text_content already has content (from delta processing),
791                                            // skip message.content to avoid sending duplicate full content
792                                            let skip_content = !text_content.is_empty();
793                                            if let Some(reasoning) = message.reasoning_content {
794                                                // Reasoning is its OWN channel — accumulate into
795                                                // reasoning_content, NEVER text_content. Merging it into
796                                                // text_content made the assistant's `content` carry the
797                                                // chain-of-thought, so a reasoning-only turn looked like a
798                                                // finished text answer (response.text() non-empty) and the
799                                                // agent loop terminated before the model emitted its tool
800                                                // call → workers never reach generate_object → asset-diagnose
801                                                // "未返回结构化输出". It also tripped `skip_content`, dropping
802                                                // the model's real content.
803                                                if first_token_ms.is_none() {
804                                                    first_token_ms = Some(
805                                                        request_started_at.elapsed().as_millis()
806                                                            as u64,
807                                                    );
808                                                }
809                                                if let Some(delta) = Self::merge_stream_text(
810                                                    &mut reasoning_content_accum,
811                                                    &reasoning,
812                                                ) {
813                                                    let _ = tx
814                                                        .send(StreamEvent::ReasoningDelta(delta))
815                                                        .await;
816                                                }
817                                            }
818                                            if !skip_content {
819                                                if let Some(content) = message
820                                                    .content
821                                                    .filter(|value| !value.is_empty())
822                                                {
823                                                    if first_token_ms.is_none() {
824                                                        first_token_ms = Some(
825                                                            request_started_at.elapsed().as_millis()
826                                                                as u64,
827                                                        );
828                                                    }
829                                                    if let Some(delta) = Self::merge_stream_text(
830                                                        &mut text_content,
831                                                        &content,
832                                                    ) {
833                                                        let _ = tx
834                                                            .send(StreamEvent::TextDelta(delta))
835                                                            .await;
836                                                    }
837                                                }
838                                            }
839                                            if let Some(tcs) = message.tool_calls {
840                                                for (index, tc) in tcs.into_iter().enumerate() {
841                                                    tool_calls.insert(
842                                                        index,
843                                                        (
844                                                            tc.id,
845                                                            tc.function.name,
846                                                            tc.function.arguments,
847                                                        ),
848                                                    );
849                                                }
850                                            }
851                                        } else if let Some(delta) = choice.delta {
852                                            if let Some(ref rc) = delta.reasoning_content {
853                                                // Reasoning stays in reasoning_content, never text_content
854                                                // (see the message-branch note above).
855                                                if first_token_ms.is_none() {
856                                                    first_token_ms = Some(
857                                                        request_started_at.elapsed().as_millis()
858                                                            as u64,
859                                                    );
860                                                }
861                                                if let Some(delta) = Self::merge_stream_text(
862                                                    &mut reasoning_content_accum,
863                                                    rc,
864                                                ) {
865                                                    let _ = tx
866                                                        .send(StreamEvent::ReasoningDelta(delta))
867                                                        .await;
868                                                }
869                                            }
870
871                                            if let Some(content) = delta.content {
872                                                if first_token_ms.is_none() {
873                                                    first_token_ms = Some(
874                                                        request_started_at.elapsed().as_millis()
875                                                            as u64,
876                                                    );
877                                                }
878                                                if let Some(delta) = Self::merge_stream_text(
879                                                    &mut text_content,
880                                                    &content,
881                                                ) {
882                                                    let _ = tx
883                                                        .send(StreamEvent::TextDelta(delta))
884                                                        .await;
885                                                }
886                                            }
887
888                                            if let Some(tcs) = delta.tool_calls {
889                                                for tc in tcs {
890                                                    let entry = tool_calls
891                                                        .entry(tc.index)
892                                                        .or_insert_with(|| {
893                                                            (
894                                                                String::new(),
895                                                                String::new(),
896                                                                String::new(),
897                                                            )
898                                                        });
899
900                                                    if let Some(id) = tc.id {
901                                                        entry.0 = id;
902                                                    }
903                                                    if let Some(func) = tc.function {
904                                                        if let Some(name) = func.name {
905                                                            if first_token_ms.is_none() {
906                                                                first_token_ms = Some(
907                                                                    request_started_at
908                                                                        .elapsed()
909                                                                        .as_millis()
910                                                                        as u64,
911                                                                );
912                                                            }
913                                                            entry.1 = name.clone();
914                                                            let _ = tx
915                                                                .send(StreamEvent::ToolUseStart {
916                                                                    id: entry.0.clone(),
917                                                                    name,
918                                                                })
919                                                                .await;
920                                                        }
921                                                        if let Some(args) = func.arguments {
922                                                            entry.2.push_str(&args);
923                                                            let _ = tx
924                                                                .send(
925                                                                    StreamEvent::ToolUseInputDelta(
926                                                                        args,
927                                                                    ),
928                                                                )
929                                                                .await;
930                                                        }
931                                                    }
932                                                }
933                                            }
934                                        }
935                                    }
936                                }
937                            }
938                        }
939                    }
940                }
941
942                if saw_done {
943                    return;
944                }
945
946                let trailing = buffer.trim();
947                if !trailing.is_empty() {
948                    if let Ok(event) = serde_json::from_str::<OpenAiStreamChunk>(trailing) {
949                        parsed_any_event = true;
950                        if response_id.is_none() {
951                            response_id = event.id.clone();
952                        }
953                        if response_model.is_none() {
954                            response_model = event.model.clone();
955                        }
956                        if response_object.is_none() {
957                            response_object = event.object.clone();
958                        }
959                        if let Some(u) = event.usage {
960                            usage.prompt_tokens = u.prompt_tokens;
961                            usage.completion_tokens = u.completion_tokens;
962                            usage.total_tokens = u.total_tokens;
963                            usage.cache_read_tokens = u
964                                .prompt_tokens_details
965                                .as_ref()
966                                .and_then(|d| d.cached_tokens);
967                        }
968                        if let Some(choice) = event.choices.into_iter().next() {
969                            if let Some(logprobs) = choice.logprobs.as_ref() {
970                                token_logprobs.extend(openai_logprobs_to_token_logprobs(logprobs));
971                            }
972                            if let Some(reason) = choice.finish_reason {
973                                finish_reason = Some(reason);
974                            }
975                            // If text_content already has content (from delta processing),
976                            // skip message.content to avoid sending duplicate full content
977                            let skip_content = !text_content.is_empty();
978                            if let Some(message) = choice.message {
979                                if let Some(reasoning) = message.reasoning_content {
980                                    // Reasoning → reasoning_content only, never text_content
981                                    // (see the note in complete_streaming).
982                                    if first_token_ms.is_none() {
983                                        first_token_ms =
984                                            Some(request_started_at.elapsed().as_millis() as u64);
985                                    }
986                                    if let Some(delta) = Self::merge_stream_text(
987                                        &mut reasoning_content_accum,
988                                        &reasoning,
989                                    ) {
990                                        let _ = tx.send(StreamEvent::ReasoningDelta(delta)).await;
991                                    }
992                                }
993                                if !skip_content {
994                                    if let Some(content) =
995                                        message.content.filter(|value| !value.is_empty())
996                                    {
997                                        if first_token_ms.is_none() {
998                                            first_token_ms = Some(
999                                                request_started_at.elapsed().as_millis() as u64,
1000                                            );
1001                                        }
1002                                        if let Some(delta) =
1003                                            Self::merge_stream_text(&mut text_content, &content)
1004                                        {
1005                                            let _ = tx.send(StreamEvent::TextDelta(delta)).await;
1006                                        }
1007                                    }
1008                                }
1009                                if let Some(tcs) = message.tool_calls {
1010                                    for (index, tc) in tcs.into_iter().enumerate() {
1011                                        tool_calls.insert(
1012                                            index,
1013                                            (tc.id, tc.function.name, tc.function.arguments),
1014                                        );
1015                                    }
1016                                }
1017                            } else if let Some(delta) = choice.delta {
1018                                if let Some(ref rc) = delta.reasoning_content {
1019                                    // Reasoning → reasoning_content only, never text_content
1020                                    // (see the note in complete_streaming).
1021                                    if first_token_ms.is_none() {
1022                                        first_token_ms =
1023                                            Some(request_started_at.elapsed().as_millis() as u64);
1024                                    }
1025                                    if let Some(delta) =
1026                                        Self::merge_stream_text(&mut reasoning_content_accum, rc)
1027                                    {
1028                                        let _ = tx.send(StreamEvent::ReasoningDelta(delta)).await;
1029                                    }
1030                                }
1031                                if let Some(content) = delta.content {
1032                                    if first_token_ms.is_none() {
1033                                        first_token_ms =
1034                                            Some(request_started_at.elapsed().as_millis() as u64);
1035                                    }
1036                                    if let Some(delta) =
1037                                        Self::merge_stream_text(&mut text_content, &content)
1038                                    {
1039                                        let _ = tx.send(StreamEvent::TextDelta(delta)).await;
1040                                    }
1041                                }
1042                            }
1043                        }
1044                    } else if let Ok(response) = serde_json::from_str::<OpenAiResponse>(trailing) {
1045                        parsed_any_event = true;
1046                        response_id = response.id.clone();
1047                        response_model = response.model.clone();
1048                        response_object = response.object.clone();
1049                        usage.prompt_tokens = response.usage.prompt_tokens;
1050                        usage.completion_tokens = response.usage.completion_tokens;
1051                        usage.total_tokens = response.usage.total_tokens;
1052                        // MiniMax: fall back to total_characters when total_tokens is 0.
1053                        if usage.total_tokens == 0 {
1054                            usage.total_tokens = response.usage.total_characters.unwrap_or(0);
1055                        }
1056                        usage.cache_read_tokens = response
1057                            .usage
1058                            .prompt_tokens_details
1059                            .as_ref()
1060                            .and_then(|d| d.cached_tokens);
1061
1062                        if let Some(choice) = response.choices.into_iter().next() {
1063                            if let Some(logprobs) = choice.logprobs.as_ref() {
1064                                token_logprobs.extend(openai_logprobs_to_token_logprobs(logprobs));
1065                            }
1066                            finish_reason = choice.finish_reason;
1067                            if let Some(text) =
1068                                choice.message.content.filter(|text| !text.is_empty())
1069                            {
1070                                if first_token_ms.is_none() {
1071                                    first_token_ms =
1072                                        Some(request_started_at.elapsed().as_millis() as u64);
1073                                }
1074                                let _ = Self::merge_stream_text(&mut text_content, &text);
1075                            }
1076                            if let Some(reasoning) = choice.message.reasoning_content {
1077                                reasoning_content_accum.push_str(&reasoning);
1078                            }
1079                            if let Some(final_tool_calls) = choice.message.tool_calls {
1080                                for tc in final_tool_calls {
1081                                    tool_calls.insert(
1082                                        tool_calls.len(),
1083                                        (tc.id, tc.function.name, tc.function.arguments),
1084                                    );
1085                                }
1086                            }
1087                        }
1088                    }
1089                }
1090
1091                if parsed_any_event
1092                    || !text_content.is_empty()
1093                    || !tool_calls.is_empty()
1094                    || !content_blocks.is_empty()
1095                {
1096                    tracing::warn!(
1097                        provider = %provider_name,
1098                        model = %request_model,
1099                        "OpenAI-compatible stream ended without [DONE]; finalizing buffered response"
1100                    );
1101                    if !text_content.is_empty() {
1102                        content_blocks.push(ContentBlock::Text {
1103                            text: text_content.clone(),
1104                        });
1105                    }
1106                    for (_, (id, name, args)) in tool_calls.iter() {
1107                        content_blocks.push(ContentBlock::ToolUse {
1108                            id: id.clone(),
1109                            name: name.clone(),
1110                            input: Self::parse_tool_arguments(name, args),
1111                        });
1112                    }
1113                    tool_calls.clear();
1114                    crate::telemetry::record_llm_usage(
1115                        usage.prompt_tokens,
1116                        usage.completion_tokens,
1117                        usage.total_tokens,
1118                        finish_reason.as_deref(),
1119                    );
1120                    let response = LlmResponse {
1121                        message: Message {
1122                            role: "assistant".to_string(),
1123                            content: std::mem::take(&mut content_blocks),
1124                            reasoning_content: if reasoning_content_accum.is_empty() {
1125                                None
1126                            } else {
1127                                Some(std::mem::take(&mut reasoning_content_accum))
1128                            },
1129                        },
1130                        usage: usage.clone(),
1131                        stop_reason: std::mem::take(&mut finish_reason),
1132                        token_logprobs: std::mem::take(&mut token_logprobs),
1133                        meta: Some(LlmResponseMeta {
1134                            provider: Some(provider_name.clone()),
1135                            request_model: Some(request_model.clone()),
1136                            request_url: Some(request_url.clone()),
1137                            response_id: response_id.clone(),
1138                            response_model: response_model.clone(),
1139                            response_object: response_object.clone(),
1140                            first_token_ms,
1141                            duration_ms: Some(request_started_at.elapsed().as_millis() as u64),
1142                        }),
1143                    };
1144                    let _ = tx.send(StreamEvent::Done(response)).await;
1145                } else {
1146                    tracing::warn!(
1147                        provider = %provider_name,
1148                        model = %request_model,
1149                        trailing = %trailing.chars().take(400).collect::<String>(),
1150                        "OpenAI-compatible stream ended without any parseable events"
1151                    );
1152                }
1153            });
1154
1155            Ok(rx)
1156        }
1157    }
1158}
1159
1160fn openai_logprobs_to_token_logprobs(logprobs: &OpenAiChoiceLogprobs) -> Vec<TokenLogProb> {
1161    logprobs
1162        .content
1163        .as_ref()
1164        .map(|items| {
1165            items
1166                .iter()
1167                .map(|item| TokenLogProb {
1168                    token: item.token.clone(),
1169                    logprob: item.logprob,
1170                    bytes: item.bytes.clone(),
1171                    top_logprobs: item
1172                        .top_logprobs
1173                        .iter()
1174                        .map(|top| TopTokenLogProb {
1175                            token: top.token.clone(),
1176                            logprob: top.logprob,
1177                            bytes: top.bytes.clone(),
1178                        })
1179                        .collect(),
1180                })
1181                .collect()
1182        })
1183        .unwrap_or_default()
1184}
1185
1186// OpenAI API response types (private)
1187#[derive(Debug, Deserialize)]
1188pub(crate) struct OpenAiResponse {
1189    #[serde(default)]
1190    pub(crate) id: Option<String>,
1191    #[serde(default)]
1192    pub(crate) object: Option<String>,
1193    #[serde(default)]
1194    pub(crate) model: Option<String>,
1195    pub(crate) choices: Vec<OpenAiChoice>,
1196    pub(crate) usage: OpenAiUsage,
1197}
1198
1199#[derive(Debug, Deserialize)]
1200pub(crate) struct OpenAiChoice {
1201    pub(crate) message: OpenAiMessage,
1202    pub(crate) finish_reason: Option<String>,
1203    #[serde(default)]
1204    pub(crate) logprobs: Option<OpenAiChoiceLogprobs>,
1205}
1206
1207#[derive(Debug, Deserialize)]
1208pub(crate) struct OpenAiChoiceLogprobs {
1209    #[serde(default)]
1210    pub(crate) content: Option<Vec<OpenAiTokenLogprob>>,
1211}
1212
1213#[derive(Debug, Deserialize)]
1214pub(crate) struct OpenAiTokenLogprob {
1215    pub(crate) token: String,
1216    pub(crate) logprob: f64,
1217    #[serde(default)]
1218    pub(crate) bytes: Option<Vec<u8>>,
1219    #[serde(default)]
1220    pub(crate) top_logprobs: Vec<OpenAiTopLogprob>,
1221}
1222
1223#[derive(Debug, Deserialize)]
1224pub(crate) struct OpenAiTopLogprob {
1225    pub(crate) token: String,
1226    pub(crate) logprob: f64,
1227    #[serde(default)]
1228    pub(crate) bytes: Option<Vec<u8>>,
1229}
1230
1231#[derive(Debug, Deserialize)]
1232pub(crate) struct OpenAiMessage {
1233    // glm5.1 (and other GLM/zhipu reasoning models) stream/return reasoning under
1234    // `reasoning`, not the `reasoning_content` kimi/deepseek use. Without this alias the
1235    // reasoning phase yields zero recognized deltas → no ReasoningDelta events → the
1236    // stream-stall watchdog kills long reasoning mid-flight (asset-diagnose "未返回结构化输出").
1237    #[serde(alias = "reasoning")]
1238    pub(crate) reasoning_content: Option<String>,
1239    pub(crate) content: Option<String>,
1240    pub(crate) tool_calls: Option<Vec<OpenAiToolCall>>,
1241}
1242
1243#[derive(Debug, Deserialize)]
1244pub(crate) struct OpenAiToolCall {
1245    pub(crate) id: String,
1246    pub(crate) function: OpenAiFunction,
1247}
1248
1249#[derive(Debug, Deserialize)]
1250pub(crate) struct OpenAiFunction {
1251    pub(crate) name: String,
1252    pub(crate) arguments: String,
1253}
1254
1255#[derive(Debug, Deserialize)]
1256pub(crate) struct OpenAiUsage {
1257    #[serde(default)]
1258    pub(crate) prompt_tokens: usize,
1259    #[serde(default)]
1260    pub(crate) completion_tokens: usize,
1261    #[serde(default)]
1262    pub(crate) total_tokens: usize,
1263    /// MiniMax uses `total_characters` instead of token counts.
1264    #[serde(default)]
1265    pub(crate) total_characters: Option<usize>,
1266    /// OpenAI returns cached token count in `prompt_tokens_details.cached_tokens`
1267    #[serde(default)]
1268    pub(crate) prompt_tokens_details: Option<OpenAiPromptTokensDetails>,
1269}
1270
1271#[derive(Debug, Deserialize)]
1272pub(crate) struct OpenAiPromptTokensDetails {
1273    #[serde(default)]
1274    pub(crate) cached_tokens: Option<usize>,
1275}
1276
1277// OpenAI streaming types
1278#[derive(Debug, Deserialize)]
1279pub(crate) struct OpenAiStreamChunk {
1280    #[serde(default)]
1281    pub(crate) id: Option<String>,
1282    #[serde(default)]
1283    pub(crate) object: Option<String>,
1284    #[serde(default)]
1285    pub(crate) model: Option<String>,
1286    pub(crate) choices: Vec<OpenAiStreamChoice>,
1287    pub(crate) usage: Option<OpenAiUsage>,
1288}
1289
1290#[derive(Debug, Deserialize)]
1291pub(crate) struct OpenAiStreamChoice {
1292    pub(crate) message: Option<OpenAiMessage>,
1293    pub(crate) delta: Option<OpenAiDelta>,
1294    pub(crate) finish_reason: Option<String>,
1295    #[serde(default)]
1296    pub(crate) logprobs: Option<OpenAiChoiceLogprobs>,
1297}
1298
1299#[derive(Debug, Deserialize)]
1300pub(crate) struct OpenAiDelta {
1301    // glm5.1 (and other GLM/zhipu reasoning models) stream/return reasoning under
1302    // `reasoning`, not the `reasoning_content` kimi/deepseek use. Without this alias the
1303    // reasoning phase yields zero recognized deltas → no ReasoningDelta events → the
1304    // stream-stall watchdog kills long reasoning mid-flight (asset-diagnose "未返回结构化输出").
1305    #[serde(alias = "reasoning")]
1306    pub(crate) reasoning_content: Option<String>,
1307    pub(crate) content: Option<String>,
1308    pub(crate) tool_calls: Option<Vec<OpenAiToolCallDelta>>,
1309}
1310
1311#[derive(Debug, Deserialize)]
1312pub(crate) struct OpenAiToolCallDelta {
1313    pub(crate) index: usize,
1314    pub(crate) id: Option<String>,
1315    pub(crate) function: Option<OpenAiFunctionDelta>,
1316}
1317
1318#[derive(Debug, Deserialize)]
1319pub(crate) struct OpenAiFunctionDelta {
1320    pub(crate) name: Option<String>,
1321    pub(crate) arguments: Option<String>,
1322}
1323
1324// ============================================================================
1325// Tests
1326// ============================================================================
1327
1328#[cfg(test)]
1329mod tests {
1330    use super::*;
1331    use crate::llm::types::{Message, ToolDefinition};
1332
1333    fn make_client() -> OpenAiClient {
1334        OpenAiClient::new("test-key".to_string(), "gpt-test".to_string())
1335    }
1336
1337    // --- streaming reasoning-channel regression -----------------------------
1338    // Reasoning models (glm5.1/zhipu) stream chain-of-thought under `reasoning`.
1339    // It must land in reasoning_content, NEVER in the text content — otherwise
1340    // response.text() looks like a finished answer and the agent loop terminates
1341    // before the model emits its tool call (asset-diagnose "未返回结构化输出").
1342
1343    struct MockSseHttp {
1344        chunks: Vec<String>,
1345    }
1346
1347    #[async_trait::async_trait]
1348    impl crate::llm::http::HttpClient for MockSseHttp {
1349        async fn post(
1350            &self,
1351            _url: &str,
1352            _headers: Vec<(&str, &str)>,
1353            _body: &serde_json::Value,
1354            _cancel: tokio_util::sync::CancellationToken,
1355        ) -> anyhow::Result<crate::llm::http::HttpResponse> {
1356            anyhow::bail!("post is unused in the streaming test")
1357        }
1358
1359        async fn post_streaming(
1360            &self,
1361            _url: &str,
1362            _headers: Vec<(&str, &str)>,
1363            _body: &serde_json::Value,
1364            _cancel: tokio_util::sync::CancellationToken,
1365        ) -> anyhow::Result<crate::llm::http::StreamingHttpResponse> {
1366            let items: Vec<anyhow::Result<bytes::Bytes>> = self
1367                .chunks
1368                .iter()
1369                .map(|s| Ok(bytes::Bytes::from(s.clone())))
1370                .collect();
1371            Ok(crate::llm::http::StreamingHttpResponse {
1372                status: 200,
1373                retry_after: None,
1374                byte_stream: Box::pin(futures::stream::iter(items)),
1375                error_body: String::new(),
1376            })
1377        }
1378    }
1379
1380    fn glm_client(chunks: Vec<String>) -> OpenAiClient {
1381        OpenAiClient::new("k".to_string(), "glm-test".to_string())
1382            .with_http_client(std::sync::Arc::new(MockSseHttp { chunks }))
1383    }
1384
1385    async fn drain_to_done(client: &OpenAiClient) -> crate::llm::LlmResponse {
1386        use crate::llm::{LlmClient, StreamEvent};
1387        let mut rx = client
1388            .complete_streaming(
1389                &[Message::user("go")],
1390                None,
1391                &[],
1392                tokio_util::sync::CancellationToken::new(),
1393            )
1394            .await
1395            .expect("stream opened");
1396        let mut done = None;
1397        while let Some(ev) = rx.recv().await {
1398            if let StreamEvent::Done(resp) = ev {
1399                done = Some(resp);
1400            }
1401        }
1402        done.expect("a Done event")
1403    }
1404
1405    #[tokio::test]
1406    async fn streaming_reasoning_does_not_leak_into_content_and_keeps_tool_call() {
1407        let chunks = vec![
1408            "data: {\"choices\":[{\"delta\":{\"reasoning\":\"Let me plan the workers\"}}]}\n\n"
1409                .to_string(),
1410            "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"parallel_task\",\"arguments\":\"{}\"}}]}}]}\n\n"
1411                .to_string(),
1412            "data: [DONE]\n\n".to_string(),
1413        ];
1414        let resp = drain_to_done(&glm_client(chunks)).await;
1415        // Reasoning must NOT appear as text content.
1416        assert_eq!(resp.message.text(), "", "reasoning leaked into content");
1417        assert_eq!(
1418            resp.message.reasoning_content.as_deref(),
1419            Some("Let me plan the workers")
1420        );
1421        // The tool call still survives, so the agent can act.
1422        let calls = resp.message.tool_calls();
1423        assert_eq!(calls.len(), 1);
1424        assert_eq!(calls[0].name, "parallel_task");
1425    }
1426
1427    #[tokio::test]
1428    async fn streaming_reasoning_only_turn_yields_empty_text() {
1429        // A pure "thinking" turn (reasoning, no content, no tool call) must yield empty
1430        // text() so the agent loop's looks_incomplete("")==true path CONTINUES instead of
1431        // terminating prematurely — the multi-worker diagnose failure root cause.
1432        let chunks = vec![
1433            "data: {\"choices\":[{\"delta\":{\"reasoning\":\"still thinking, no answer yet\"}}]}\n\n"
1434                .to_string(),
1435            "data: [DONE]\n\n".to_string(),
1436        ];
1437        let resp = drain_to_done(&glm_client(chunks)).await;
1438        assert_eq!(resp.message.text(), "");
1439        assert_eq!(
1440            resp.message.reasoning_content.as_deref(),
1441            Some("still thinking, no answer yet")
1442        );
1443        assert!(resp.message.tool_calls().is_empty());
1444    }
1445
1446    #[tokio::test]
1447    async fn streaming_collects_token_logprobs() {
1448        let chunks = vec![
1449            "data: {\"choices\":[{\"delta\":{\"content\":\"hello\"},\"logprobs\":{\"content\":[{\"token\":\"hello\",\"logprob\":-0.2,\"bytes\":[104,101,108,108,111],\"top_logprobs\":[{\"token\":\"hi\",\"logprob\":-1.2,\"bytes\":[104,105]}]}]}}]}\n\n"
1450                .to_string(),
1451            "data: {\"choices\":[{\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1,\"total_tokens\":2}}\n\n"
1452                .to_string(),
1453            "data: [DONE]\n\n".to_string(),
1454        ];
1455        let resp = drain_to_done(&glm_client(chunks).with_logprobs(true)).await;
1456        assert_eq!(resp.text(), "hello");
1457        assert_eq!(resp.token_logprobs.len(), 1);
1458        assert_eq!(resp.token_logprobs[0].token, "hello");
1459        assert_eq!(resp.token_logprobs[0].logprob, -0.2);
1460        assert_eq!(
1461            resp.token_logprobs[0].bytes.as_deref(),
1462            Some(&[104, 101, 108, 108, 111][..])
1463        );
1464        assert_eq!(resp.token_logprobs[0].top_logprobs[0].token, "hi");
1465        assert_eq!(resp.token_logprobs[0].top_logprobs[0].logprob, -1.2);
1466    }
1467
1468    #[tokio::test]
1469    async fn streaming_accepts_sse_data_without_space_after_colon() {
1470        let chunks = vec![
1471            "data:{\"choices\":[{\"delta\":{\"content\":\"hello\"},\"finish_reason\":null}],\"usage\":null}\n\n"
1472                .to_string(),
1473            "data:{\"choices\":[{\"delta\":{},\"finish_reason\":\"stop\"}],\"usage\":{\"prompt_tokens\":1,\"completion_tokens\":1,\"total_tokens\":2}}\n\n"
1474                .to_string(),
1475            "data:[DONE]\n\n".to_string(),
1476        ];
1477        let resp = drain_to_done(&glm_client(chunks)).await;
1478        assert_eq!(resp.text(), "hello");
1479        assert_eq!(resp.usage.prompt_tokens, 1);
1480        assert_eq!(resp.usage.completion_tokens, 1);
1481        assert_eq!(resp.usage.total_tokens, 2);
1482        assert_eq!(resp.stop_reason.as_deref(), Some("stop"));
1483    }
1484
1485    #[test]
1486    fn test_apply_directive_forced_function_tool_choice() {
1487        let mut req = serde_json::json!({ "model": "m" });
1488        OpenAiClient::apply_directive(
1489            &mut req,
1490            &structured::StructuredDirective {
1491                force_tool: Some("emit_person".to_string()),
1492                response_format: None,
1493            },
1494        );
1495        assert_eq!(req["tool_choice"]["type"], "function");
1496        assert_eq!(req["tool_choice"]["function"]["name"], "emit_person");
1497        assert!(req.get("response_format").is_none());
1498    }
1499
1500    #[test]
1501    fn test_apply_directive_json_schema_strict() {
1502        let mut req = serde_json::json!({});
1503        OpenAiClient::apply_directive(
1504            &mut req,
1505            &structured::StructuredDirective {
1506                force_tool: None,
1507                response_format: Some(structured::ResponseFormat::JsonSchema {
1508                    name: "person".to_string(),
1509                    schema: serde_json::json!({ "type": "object" }),
1510                }),
1511            },
1512        );
1513        assert_eq!(req["response_format"]["type"], "json_schema");
1514        assert_eq!(req["response_format"]["json_schema"]["name"], "person");
1515        assert_eq!(req["response_format"]["json_schema"]["strict"], true);
1516        assert!(req.get("tool_choice").is_none());
1517    }
1518
1519    #[test]
1520    fn test_apply_directive_json_object() {
1521        let mut req = serde_json::json!({});
1522        OpenAiClient::apply_directive(
1523            &mut req,
1524            &structured::StructuredDirective {
1525                force_tool: None,
1526                response_format: Some(structured::ResponseFormat::JsonObject),
1527            },
1528        );
1529        assert_eq!(req["response_format"]["type"], "json_object");
1530    }
1531
1532    #[test]
1533    fn test_build_chat_request_applies_directive_and_system() {
1534        let req = make_client().build_chat_request(
1535            &[Message::user("hi")],
1536            Some("sys"),
1537            &[ToolDefinition {
1538                name: "emit_x".to_string(),
1539                description: "emit".to_string(),
1540                parameters: serde_json::json!({ "type": "object" }),
1541            }],
1542            Some(&structured::StructuredDirective {
1543                force_tool: Some("emit_x".to_string()),
1544                response_format: None,
1545            }),
1546        );
1547        assert_eq!(req["messages"][0]["role"], "system");
1548        assert_eq!(req["tool_choice"]["function"]["name"], "emit_x");
1549        assert_eq!(req["tools"][0]["function"]["name"], "emit_x");
1550    }
1551
1552    #[test]
1553    fn test_build_chat_request_without_directive_is_plain() {
1554        let req = make_client().build_chat_request(&[Message::user("hi")], None, &[], None);
1555        assert!(req.get("tool_choice").is_none());
1556        assert!(req.get("response_format").is_none());
1557        assert!(req.get("logprobs").is_none());
1558        assert!(req.get("top_logprobs").is_none());
1559    }
1560
1561    #[test]
1562    fn test_build_chat_request_includes_logprob_options_when_enabled() {
1563        let req = make_client().with_top_logprobs(1).build_chat_request(
1564            &[Message::user("hi")],
1565            None,
1566            &[],
1567            None,
1568        );
1569        assert_eq!(req["logprobs"], true);
1570        assert_eq!(req["top_logprobs"], 1);
1571    }
1572
1573    #[test]
1574    fn test_parse_openai_token_logprobs() {
1575        let parsed = openai_logprobs_to_token_logprobs(&OpenAiChoiceLogprobs {
1576            content: Some(vec![OpenAiTokenLogprob {
1577                token: "hello".to_string(),
1578                logprob: -0.25,
1579                bytes: Some(vec![104, 101, 108, 108, 111]),
1580                top_logprobs: vec![OpenAiTopLogprob {
1581                    token: "hi".to_string(),
1582                    logprob: -1.5,
1583                    bytes: Some(vec![104, 105]),
1584                }],
1585            }]),
1586        });
1587        assert_eq!(parsed.len(), 1);
1588        assert_eq!(parsed[0].token, "hello");
1589        assert_eq!(parsed[0].logprob, -0.25);
1590        assert_eq!(
1591            parsed[0].bytes.as_deref(),
1592            Some(&[104, 101, 108, 108, 111][..])
1593        );
1594        assert_eq!(parsed[0].top_logprobs[0].token, "hi");
1595        assert_eq!(parsed[0].top_logprobs[0].logprob, -1.5);
1596    }
1597
1598    #[test]
1599    fn test_native_structured_support_is_json_schema() {
1600        assert_eq!(
1601            make_client().native_structured_support(),
1602            structured::NativeStructuredSupport::JsonSchema
1603        );
1604    }
1605
1606    #[test]
1607    fn test_native_structured_support_can_be_overridden() {
1608        assert_eq!(
1609            make_client()
1610                .with_native_structured_support(structured::NativeStructuredSupport::None)
1611                .native_structured_support(),
1612            structured::NativeStructuredSupport::None
1613        );
1614    }
1615}
a3s_code_core/llm/openai.rs

a3s_code_core/llm/
openai.rs
