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

a3s_code_core/llm/
openai.rs
