diff --git a/crates/dbx-core/src/agent_loop.rs b/crates/dbx-core/src/agent_loop.rs index 34f710af51..48882add99 100644 --- a/crates/dbx-core/src/agent_loop.rs +++ b/crates/dbx-core/src/agent_loop.rs @@ -1020,6 +1020,12 @@ fn context_window_for_model(model: &str) -> u32 { if m.contains("gpt-4.1") { return 1_000_000; } + if m.contains("minimax-m3") { + return 1_000_000; + } + if m.contains("minimax-m2") { + return 204_800; + } if m.contains("claude") || m.contains("o1") || m.starts_with("o3") || m.starts_with("o4") { 200_000 } else if m.contains("gpt-4") { @@ -1031,6 +1037,10 @@ fn context_window_for_model(model: &str) -> u32 { } } +fn effective_context_window(config: &AiConfig) -> u32 { + config.context_window.unwrap_or_else(|| context_window_for_model(&config.model)) +} + fn prompt_budget(window: u32, max_tokens: Option) -> u32 { let output_reserve = max_tokens.unwrap_or(4096).min(window / 2); let safety_reserve = (window / 10).clamp(2048, 16_384).min(window / 2); @@ -1061,7 +1071,7 @@ async fn maybe_compact( cancelled: &Notify, force: bool, ) -> CompactResult { - let window = config.context_window.unwrap_or_else(|| context_window_for_model(&config.model)); + let window = effective_context_window(config); let budget = prompt_budget(window, max_tokens); let estimated_before = estimate_current_prompt_tokens(system_prompt, tools, messages); @@ -1353,6 +1363,27 @@ mod tests { assert_eq!(clamp_max_agent_turns(u32::MAX), MAX_MAX_AGENT_TURNS); } + #[test] + fn minimax_context_windows_follow_official_model_families() { + assert_eq!(context_window_for_model("MiniMax-M3"), 1_000_000); + assert_eq!(context_window_for_model("vendor/MiniMax-M3.1"), 1_000_000); + assert_eq!(context_window_for_model("MiniMax-M2.7"), 204_800); + assert_eq!(context_window_for_model("MiniMax-M2.5-highspeed"), 204_800); + assert_eq!(context_window_for_model("MiniMax-future"), 128_000); + } + + #[test] + fn explicit_context_window_overrides_minimax_family_default() { + let config: AiConfig = serde_json::from_value(serde_json::json!({ + "provider": "minimax", + "model": "MiniMax-M3", + "contextWindow": 65_536 + })) + .unwrap(); + + assert_eq!(effective_context_window(&config), 65_536); + } + #[test] fn identifies_explicit_tool_rejection_errors() { for error in [ diff --git a/crates/dbx-core/src/ai.rs b/crates/dbx-core/src/ai.rs index 1ad7b01d6a..225bcc2a2a 100644 --- a/crates/dbx-core/src/ai.rs +++ b/crates/dbx-core/src/ai.rs @@ -845,6 +845,207 @@ pub fn openai_stream_reasoning(event: &serde_json::Value) -> Option<&str> { .filter(|text| !text.is_empty()) } +const MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: &str = "minimax_reasoning_details"; + +#[derive(Debug, Clone, Copy, Default)] +enum MiniMaxStreamSemantics { + #[default] + Auto, + Incremental, +} + +fn minimax_stream_semantics(config: &AiConfig) -> MiniMaxStreamSemantics { + // The China platform currently emits ordinary fragments, while the global + // SDK example documents cumulative snapshots. A fixed regional mode avoids + // the inherently ambiguous case where two incremental fragments happen to + // share a prefix; custom gateways retain the tolerant auto mode. + let china_endpoint = reqwest::Url::parse(&config.endpoint) + .ok() + .and_then(|url| url.host_str().map(ToString::to_string)) + .is_some_and(|host| host.eq_ignore_ascii_case("api.minimaxi.com")); + if china_endpoint { + MiniMaxStreamSemantics::Incremental + } else { + MiniMaxStreamSemantics::Auto + } +} + +fn minimax_stream_reasoning_details(event: &serde_json::Value) -> Option<&Vec> { + event["choices"] + .get(0) + .and_then(|choice| choice["delta"]["reasoning_details"].as_array()) + .filter(|details| !details.is_empty()) +} + +/// Converts cumulative text snapshots or ordinary incremental chunks into +/// incremental deltas while retaining the complete assembled text. +/// +/// MiniMax deployments do not all use the same streaming semantics: the global +/// API examples use cumulative snapshots, while the China API currently emits +/// incremental fragments. Accepting both forms keeps rendering and tool-call +/// replay consistent across the official endpoints and compatible gateways. +#[derive(Debug, Default)] +struct MiniMaxTextAccumulator { + semantics: MiniMaxStreamSemantics, + latest: String, + complete: String, +} + +impl MiniMaxTextAccumulator { + fn new(semantics: MiniMaxStreamSemantics) -> Self { + Self { semantics, latest: String::new(), complete: String::new() } + } + + fn push(&mut self, value: &str) -> Option { + if value.is_empty() { + return None; + } + + if matches!(self.semantics, MiniMaxStreamSemantics::Incremental) { + self.latest = value.to_string(); + self.complete.push_str(value); + return Some(value.to_string()); + } + + if value == self.latest { + return None; + } + + if let Some(suffix) = value.strip_prefix(&self.latest) { + let delta = suffix.to_string(); + self.latest = value.to_string(); + self.complete.push_str(&delta); + return (!delta.is_empty()).then_some(delta); + } + + if self.latest.starts_with(value) { + // Ignore an older/shorter cumulative snapshot that arrived late. + return None; + } + + self.latest = value.to_string(); + self.complete.push_str(value); + Some(value.to_string()) + } + + fn complete(&self) -> &str { + &self.complete + } +} + +#[derive(Debug)] +struct MiniMaxReasoningDetailState { + position: usize, + latest: serde_json::Value, + text: MiniMaxTextAccumulator, +} + +impl MiniMaxReasoningDetailState { + fn matches(&self, detail: &serde_json::Value, position: usize) -> bool { + let id = detail["id"].as_str().filter(|id| !id.is_empty()); + if let (Some(id), Some(latest_id)) = (id, self.latest["id"].as_str().filter(|id| !id.is_empty())) { + return id == latest_id; + } + + let index = detail["index"].as_u64(); + if let (Some(index), Some(latest_index)) = (index, self.latest["index"].as_u64()) { + return index == latest_index; + } + + self.position == position + } + + fn process(&mut self, detail: &serde_json::Value) -> Option { + let delta = detail["text"].as_str().and_then(|text| self.text.push(text)); + self.latest = detail.clone(); + delta + } + + fn replay_value(&self) -> serde_json::Value { + let mut detail = self.latest.clone(); + if detail.get("text").is_some() { + detail["text"] = serde_json::Value::String(self.text.complete().to_string()); + } + detail + } +} + +#[derive(Debug)] +struct MiniMaxStreamState { + semantics: MiniMaxStreamSemantics, + content: MiniMaxTextAccumulator, + reasoning_fallback: MiniMaxTextAccumulator, + reasoning_details: Vec, +} + +#[derive(Debug, Default, PartialEq)] +struct MiniMaxStreamDelta { + text: Option, + reasoning: Option, +} + +impl MiniMaxStreamState { + fn new(semantics: MiniMaxStreamSemantics) -> Self { + Self { + semantics, + content: MiniMaxTextAccumulator::new(semantics), + reasoning_fallback: MiniMaxTextAccumulator::new(semantics), + reasoning_details: Vec::new(), + } + } + + fn process(&mut self, event: &serde_json::Value) -> MiniMaxStreamDelta { + let text = openai_stream_text(event).and_then(|text| self.content.push(&text)); + let reasoning = if let Some(details) = minimax_stream_reasoning_details(event) { + let fallback_delta = + openai_stream_reasoning(event).and_then(|reasoning| self.reasoning_fallback.push(reasoning)); + let has_detail_text = + details.iter().any(|detail| detail["text"].as_str().is_some_and(|text| !text.is_empty())); + let mut delta = String::new(); + for (position, detail) in details.iter().enumerate() { + let state_index = + self.reasoning_details.iter().position(|state| state.matches(detail, position)).unwrap_or_else( + || { + self.reasoning_details.push(MiniMaxReasoningDetailState { + position, + latest: detail.clone(), + text: MiniMaxTextAccumulator::new(self.semantics), + }); + self.reasoning_details.len() - 1 + }, + ); + if let Some(fragment) = self.reasoning_details[state_index].process(detail) { + delta.push_str(&fragment); + } + } + if has_detail_text { + (!delta.is_empty()).then_some(delta) + } else { + fallback_delta + } + } else { + openai_stream_reasoning(event).and_then(|reasoning| self.reasoning_fallback.push(reasoning)) + }; + MiniMaxStreamDelta { text, reasoning } + } + + fn provider_payload(&self) -> Option { + (!self.reasoning_details.is_empty()).then(|| { + let details = + self.reasoning_details.iter().map(MiniMaxReasoningDetailState::replay_value).collect::>(); + json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: details, + }) + }) + } +} + +impl Default for MiniMaxStreamState { + fn default() -> Self { + Self::new(MiniMaxStreamSemantics::Auto) + } +} + fn openai_stream_has_finish_reason(event: &serde_json::Value) -> bool { event["choices"].as_array().is_some_and(|choices| { choices.iter().any(|choice| choice["finish_reason"].as_str().is_some_and(|reason| !reason.is_empty())) @@ -888,18 +1089,32 @@ fn is_openai_reasoning_model(model: &str) -> bool { model.starts_with("gpt-5") || model.starts_with("o1") || model.starts_with("o3") || model.starts_with("o4") } -fn uses_openai_max_completion_tokens(config: &AiConfig) -> bool { - is_openai_api_config(config) && is_openai_reasoning_model(&config.model) +fn uses_chat_completion_max_completion_tokens(config: &AiConfig) -> bool { + matches!(config.provider, AiProvider::MiniMax) + || is_openai_api_config(config) && is_openai_reasoning_model(&config.model) } fn set_chat_completion_token_limit(body: &mut serde_json::Value, config: &AiConfig, max_tokens: u32) { - if uses_openai_max_completion_tokens(config) { + if uses_chat_completion_max_completion_tokens(config) { body["max_completion_tokens"] = json!(max_tokens); } else { body["max_tokens"] = json!(max_tokens); } } +fn apply_minimax_chat_completion_fields(body: &mut serde_json::Value, config: &AiConfig) { + if matches!(config.provider, AiProvider::MiniMax) { + body["reasoning_split"] = json!(true); + } +} + +fn decorate_chat_completion_body(body: &mut serde_json::Value, config: &AiConfig, max_tokens: u32) { + set_chat_completion_token_limit(body, config, max_tokens); + apply_minimax_chat_completion_fields(body, config); + apply_chat_completion_thinking_toggle(body, config); + crate::ai_effort::apply_runtime_effort(body, config); +} + /// Kimi K2.5+ models (including K2.7-Code) handle thinking flags differently /// and reject the OpenAI-compatible `extra_body.chat_template_kwargs` toggle. /// @@ -1131,6 +1346,10 @@ fn validate_config(config: &AiConfig) -> Result<(), String> { if is_cli_provider(&config.provider) { return Ok(()); } + if matches!(config.provider, AiProvider::MiniMax) && config.api_style != AiApiStyle::Completions { + return Err("MiniMax currently supports the Chat Completions API style in DBX; select Completions and retry" + .to_string()); + } if provider_requires_api_key(&config.provider) && config.api_key.trim().is_empty() { return Err("API key is required".to_string()); } @@ -1661,19 +1880,60 @@ pub async fn call_claude(client: &reqwest::Client, request: AiCompletionRequest) .to_string()) } +fn minimax_reasoning_details_from_tool_calls(tool_calls: &[ToolCallRef]) -> Option { + tool_calls.iter().filter_map(|tool_call| tool_call.provider_payload.as_ref()).find_map(|payload| { + payload.get(MINIMAX_REASONING_DETAILS_PAYLOAD_KEY).filter(|details| details.is_array()).cloned() + }) +} + +fn build_openai_chat_messages( + config: &AiConfig, + system_prompt: &str, + messages: &[AiMessage], +) -> Vec { + let mut output = vec![json!({ "role": "system", "content": system_prompt })]; + output.extend(messages.iter().map(|message| { + let mut item = json!({ "role": message.role, "content": message.content }); + if message.role == "tool" { + if let Some(tool_call_id) = message.tool_call_id.as_ref() { + item["tool_call_id"] = json!(tool_call_id); + } + } else if message.role == "assistant" && !message.tool_calls.is_empty() { + item["tool_calls"] = json!(message + .tool_calls + .iter() + .map(|tool_call| { + json!({ + "id": tool_call.id, + "type": "function", + "function": { + "name": tool_call.name, + "arguments": tool_call.arguments.to_string(), + } + }) + }) + .collect::>()); + if matches!(config.provider, AiProvider::MiniMax) { + if let Some(reasoning_details) = minimax_reasoning_details_from_tool_calls(&message.tool_calls) { + item["reasoning_details"] = reasoning_details; + } + } + } + item + })); + output +} + pub async fn call_openai_compatible(client: &reqwest::Client, request: AiCompletionRequest) -> Result { let headers = maybe_bearer_headers(&request.config)?; - let mut messages = vec![json!({ "role": "system", "content": request.system_prompt })]; - messages.extend(request.messages.iter().map(|message| json!({ "role": message.role, "content": message.content }))); + let messages = build_openai_chat_messages(&request.config, &request.system_prompt, &request.messages); let mut body_obj = json!({ "model": request.config.model, "messages": messages, }); - set_chat_completion_token_limit(&mut body_obj, &request.config, request.max_tokens.unwrap_or(2048)); - apply_chat_completion_thinking_toggle(&mut body_obj, &request.config); - crate::ai_effort::apply_runtime_effort(&mut body_obj, &request.config); + decorate_chat_completion_body(&mut body_obj, &request.config, request.max_tokens.unwrap_or(2048)); let res = client .post(resolve_endpoint(&request.config)) @@ -1944,6 +2204,8 @@ fn probe_stream_payload( event_name: Option<&str>, is_claude: bool, is_gemini: bool, + is_minimax: bool, + minimax_state: &mut MiniMaxStreamState, diagnostics: &mut StreamProbeDiagnostics, ) -> Result, String> { diagnostics.data_events += 1; @@ -1963,7 +2225,10 @@ fn probe_stream_payload( diagnostics.observe_gemini(&parsed); } - let delta = if is_claude { + let delta = if is_minimax { + let delta = minimax_state.process(&parsed); + delta.text.or(delta.reasoning) + } else if is_claude { claude_stream_text(&parsed).or_else(|| parsed["delta"]["thinking"].as_str()).map(ToString::to_string) } else if is_gemini { let text = gemini_text(&parsed); @@ -2046,10 +2311,12 @@ async fn measure_first_stream_chunk( start: std::time::Instant, is_claude: bool, is_gemini: bool, + is_minimax: bool, ) -> Result<(u64, String), String> { let mut buf = Vec::new(); let mut diagnostics = StreamProbeDiagnostics::default(); let mut event_name: Option = None; + let mut minimax_state = MiniMaxStreamState::default(); while let Some(chunk) = byte_stream.next().await { let chunk = chunk.map_err(|e| format!("stream read error: {}", e.without_url()))?; diagnostics.bytes_received += chunk.len(); @@ -2069,9 +2336,15 @@ async fn measure_first_stream_chunk( diagnostics.data_events += 1; return Err(diagnostics.empty_stream_error(is_gemini)); } - if let Some(text) = - probe_stream_payload(data, event_name.as_deref(), is_claude, is_gemini, &mut diagnostics)? - { + if let Some(text) = probe_stream_payload( + data, + event_name.as_deref(), + is_claude, + is_gemini, + is_minimax, + &mut minimax_state, + &mut diagnostics, + )? { let latency = start.elapsed().as_millis() as u64; return Ok((latency, text)); } @@ -2081,9 +2354,15 @@ async fn measure_first_stream_chunk( if !buf.is_empty() { let line = String::from_utf8(buf).map_err(|e| format!("AI stream returned invalid UTF-8: {e}"))?; if let Some(data) = stream_data_payload(&line) { - if let Some(text) = - probe_stream_payload(data, event_name.as_deref(), is_claude, is_gemini, &mut diagnostics)? - { + if let Some(text) = probe_stream_payload( + data, + event_name.as_deref(), + is_claude, + is_gemini, + is_minimax, + &mut minimax_state, + &mut diagnostics, + )? { let latency = start.elapsed().as_millis() as u64; return Ok((latency, text)); } @@ -2158,6 +2437,7 @@ pub async fn test_connection_core(config: &AiConfig) -> Result Result unreachable!(), _ => { - let mut body_obj = if api_style == AiApiStyle::Responses { + let body_obj = if api_style == AiApiStyle::Responses { json!({ "model": &model, "input": [{ "role": "user", "content": TEST_PROMPT }], @@ -2232,12 +2512,9 @@ pub async fn test_connection_core(config: &AiConfig) -> Result Result Ok(AiTestConnectionResult { success: true, message: format!("OK — {}ms", latency), @@ -2714,20 +2991,19 @@ async fn stream_openai( ) -> Result<(), String> { let headers = maybe_bearer_headers(&request.config)?; - let mut messages = vec![json!({ "role": "system", "content": request.system_prompt })]; - messages.extend(request.messages.iter().map(|m| json!({ "role": m.role, "content": m.content }))); + let messages = build_openai_chat_messages(&request.config, &request.system_prompt, &request.messages); let mut body_obj = json!({ "model": request.config.model, "messages": messages, "stream": true, }); - set_chat_completion_token_limit(&mut body_obj, &request.config, request.max_tokens.unwrap_or(2048)); - apply_chat_completion_thinking_toggle(&mut body_obj, &request.config); - crate::ai_effort::apply_runtime_effort(&mut body_obj, &request.config); + decorate_chat_completion_body(&mut body_obj, &request.config, request.max_tokens.unwrap_or(2048)); let endpoint = resolve_endpoint(&request.config); let config = request.config.clone(); + let is_minimax = matches!(request.config.provider, AiProvider::MiniMax); + let minimax_semantics = minimax_stream_semantics(&request.config); let emitted = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); with_stream_retry(&config, &emitted, Some(cancelled), || { @@ -2737,6 +3013,7 @@ async fn stream_openai( let session_id = session_id.to_string(); let emitted = emitted.clone(); async move { + let mut minimax_state = MiniMaxStreamState::new(minimax_semantics); let res = client .post(&endpoint) .headers(headers) @@ -2768,23 +3045,45 @@ async fn stream_openai( } if let Ok(event) = serde_json::from_str::(data) { - if let Some(reasoning) = openai_stream_reasoning(&event) { - emitted.store(true, std::sync::atomic::Ordering::Relaxed); - on_chunk(AiStreamChunk { - session_id: session_id.clone(), - delta: String::new(), - reasoning_delta: Some(reasoning.to_string()), - done: false, - }); - } - if let Some(text) = openai_stream_text(&event) { - emitted.store(true, std::sync::atomic::Ordering::Relaxed); - on_chunk(AiStreamChunk { - session_id: session_id.clone(), - delta: text, - reasoning_delta: None, - done: false, - }); + if is_minimax { + let delta = minimax_state.process(&event); + if let Some(reasoning) = delta.reasoning { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: String::new(), + reasoning_delta: Some(reasoning), + done: false, + }); + } + if let Some(text) = delta.text { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: text, + reasoning_delta: None, + done: false, + }); + } + } else { + if let Some(reasoning) = openai_stream_reasoning(&event) { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: String::new(), + reasoning_delta: Some(reasoning.to_string()), + done: false, + }); + } + if let Some(text) = openai_stream_text(&event) { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: text, + reasoning_delta: None, + done: false, + }); + } } if finish_reason_deadline.is_none() && openai_stream_has_finish_reason(&event) { finish_reason_deadline = @@ -3334,32 +3633,7 @@ async fn stream_openai_with_tools( ) -> Result, String> { let headers = maybe_bearer_headers(&request.config)?; - let mut messages = vec![json!({ "role": "system", "content": request.system_prompt })]; - messages.extend(request.messages.iter().map(|m| { - let mut msg = json!({ "role": m.role, "content": m.content }); - if m.role == "tool" { - if let Some(ref tc_id) = m.tool_call_id { - msg["tool_call_id"] = json!(tc_id); - } - } else if m.role == "assistant" && !m.tool_calls.is_empty() { - let calls: Vec = m - .tool_calls - .iter() - .map(|tc| { - json!({ - "id": tc.id, - "type": "function", - "function": { - "name": tc.name, - "arguments": tc.arguments.to_string() - } - }) - }) - .collect(); - msg["tool_calls"] = json!(calls); - } - msg - })); + let messages = build_openai_chat_messages(&request.config, &request.system_prompt, &request.messages); let tool_json: Vec = tools.iter().map(|t| t.to_openai_tool()).collect(); @@ -3372,9 +3646,10 @@ async fn stream_openai_with_tools( "stream_options": { "include_usage": true }, }); set_chat_completion_token_limit(&mut body, &request.config, request.max_tokens.unwrap_or(4096)); + apply_minimax_chat_completion_fields(&mut body, &request.config); if request.config.runtime_effort.is_none() && !request.config.enable_thinking { - if matches!(request.config.provider, AiProvider::Ollama) { + if matches!(request.config.provider, AiProvider::MiniMax | AiProvider::Ollama) { apply_chat_completion_thinking_toggle(&mut body, &request.config); } else if matches!(request.config.provider, AiProvider::Deepseek) { // DeepSeek uses its own thinking field for tool-enabled requests. @@ -3385,6 +3660,8 @@ async fn stream_openai_with_tools( let endpoint = resolve_endpoint(&request.config); let config = request.config.clone(); + let is_minimax = matches!(request.config.provider, AiProvider::MiniMax); + let minimax_semantics = minimax_stream_semantics(&request.config); let emitted = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); with_stream_retry(&config, &emitted, Some(cancelled), || { @@ -3394,6 +3671,8 @@ async fn stream_openai_with_tools( let session_id = session_id.to_string(); let emitted = emitted.clone(); async move { + let mut minimax_state = MiniMaxStreamState::new(minimax_semantics); + let mut first_tool_index = None; let res = client .post(&endpoint) .headers(headers) @@ -3436,24 +3715,45 @@ async fn stream_openai_with_tools( } } // Reasoning - if let Some(reasoning) = openai_stream_reasoning(&event) { - emitted.store(true, std::sync::atomic::Ordering::Relaxed); - on_event(StreamToolEvent::Chunk(AiStreamChunk { - session_id: session_id.clone(), - delta: String::new(), - reasoning_delta: Some(reasoning.to_string()), - done: false, - })); - } - // Text - if let Some(text) = openai_stream_text(&event) { - emitted.store(true, std::sync::atomic::Ordering::Relaxed); - on_event(StreamToolEvent::Chunk(AiStreamChunk { - session_id: session_id.clone(), - delta: text, - reasoning_delta: None, - done: false, - })); + if is_minimax { + let delta = minimax_state.process(&event); + if let Some(reasoning) = delta.reasoning { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_event(StreamToolEvent::Chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: String::new(), + reasoning_delta: Some(reasoning), + done: false, + })); + } + if let Some(text) = delta.text { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_event(StreamToolEvent::Chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: text, + reasoning_delta: None, + done: false, + })); + } + } else { + if let Some(reasoning) = openai_stream_reasoning(&event) { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_event(StreamToolEvent::Chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: String::new(), + reasoning_delta: Some(reasoning.to_string()), + done: false, + })); + } + if let Some(text) = openai_stream_text(&event) { + emitted.store(true, std::sync::atomic::Ordering::Relaxed); + on_event(StreamToolEvent::Chunk(AiStreamChunk { + session_id: session_id.clone(), + delta: text, + reasoning_delta: None, + done: false, + })); + } } // Tool calls if let Some(tool_calls) = event["choices"].get(0).and_then(|c| c["delta"]["tool_calls"].as_array()) { @@ -3464,6 +3764,7 @@ async fn stream_openai_with_tools( // (e.g. GLM) send id="" on subsequent deltas, so // only a non-empty id marks a genuine start. if let Some(id) = tc["id"].as_str().filter(|s| !s.is_empty()) { + first_tool_index.get_or_insert(idx); let name = tc["function"]["name"].as_str().unwrap_or_default().to_string(); emitted.store(true, std::sync::atomic::Ordering::Relaxed); on_event(StreamToolEvent::ToolCallStart { index: idx, id: id.to_string(), name }); @@ -3496,6 +3797,10 @@ async fn stream_openai_with_tools( } } + if let (Some(index), Some(payload)) = (first_tool_index, minimax_state.provider_payload()) { + on_event(StreamToolEvent::ToolCallProviderPayload { index, payload }); + } + Ok(token_usage) } }) @@ -3970,21 +4275,24 @@ mod tests { use super::{ append_gemini_model_parts, apply_chat_completion_thinking_toggle, build_ai_http_client, build_gemini_contents, - build_responses_input_with_tools, call_claude, classify_error, claude_headers, claude_system_prompt, complete, + build_openai_chat_messages, build_responses_input_with_tools, call_claude, call_openai_compatible, + classify_error, claude_headers, claude_system_prompt, complete, decorate_chat_completion_body, drain_next_stream_line, emit_gemini_tool_call_parts, emit_responses_function_call_item, format_transport_error, gemini_text, is_kimi_model, is_retryable_error, list_models_core, maybe_bearer_headers, maybe_tag_retry_after, - measure_first_stream_chunk, merge_global_max_retries, ollama_selected_model_tool_support, openai_response_text, - openai_stream_reasoning, openai_stream_text, parse_dynamic_effort_capability, parse_gemini_model_list_response, - parse_model_list_response, parse_retry_after, parse_retry_after_secs, provider_requires_api_key, - resolve_endpoint, resolve_gemini_stream_endpoint, resolve_model_effort_core, resolve_model_list_endpoint, + measure_first_stream_chunk, merge_global_max_retries, minimax_stream_semantics, + ollama_selected_model_tool_support, openai_response_text, openai_stream_reasoning, openai_stream_text, + parse_dynamic_effort_capability, parse_gemini_model_list_response, parse_model_list_response, + parse_retry_after, parse_retry_after_secs, provider_requires_api_key, resolve_endpoint, + resolve_gemini_stream_endpoint, resolve_model_effort_core, resolve_model_list_endpoint, resolve_ollama_show_endpoint, responses_function_tool, responses_max_output_tokens, responses_stream_text, responses_text, responses_token_usage, retain_ollama_completion_models, retry_after_secs, set_chat_completion_token_limit, stream, stream_claude, stream_claude_with_tools, stream_data_payload, stream_error, stream_openai_with_tools, stream_with_tools, test_connection_core, uses_anthropic_messages_api, validate_config, validate_model_list_config, with_retry, with_stream_retry, AiApiStyle, AiAssistantMode, AiAuthMethod, AiCapabilitySource, AiChatSelectionState, AiCompletionRequest, AiConfig, AiEffortCapability, - AiEffortOption, AiEffortSelection, AiMessage, AiModelInfo, AiProvider, AiReasoningLevel, StreamToolEvent, - StreamingToolCallAccumulator, ToolCallRef, AUTHORIZATION, CLAUDE_DEFAULT_SYSTEM, TEST_PROMPT, + AiEffortOption, AiEffortSelection, AiMessage, AiModelInfo, AiProvider, AiReasoningLevel, MiniMaxStreamDelta, + MiniMaxStreamState, MiniMaxTextAccumulator, StreamToolEvent, StreamingToolCallAccumulator, ToolCallRef, + AUTHORIZATION, CLAUDE_DEFAULT_SYSTEM, MINIMAX_REASONING_DETAILS_PAYLOAD_KEY, TEST_PROMPT, }; #[test] @@ -4266,6 +4574,32 @@ mod tests { request } + fn minimax_test_config(endpoint: impl Into) -> AiConfig { + let mut config = test_config(AiProvider::MiniMax); + config.api_key = "minimax-test-key".to_string(); + config.auth_method = AiAuthMethod::Bearer; + config.endpoint = endpoint.into(); + config.model = "MiniMax-M3".to_string(); + config.api_style = AiApiStyle::Completions; + config.enable_thinking = true; + config + } + + fn minimax_test_request(endpoint: impl Into) -> AiCompletionRequest { + AiCompletionRequest { + config: minimax_test_config(endpoint), + system_prompt: "Be concise.".to_string(), + messages: vec![AiMessage { + role: "user".to_string(), + content: "Hello".to_string(), + tool_call_id: None, + tool_calls: Vec::new(), + }], + task_contract: None, + max_tokens: Some(64), + } + } + fn assert_claude_http_request(captured: CapturedJsonRequest) { assert_eq!(captured.body["model"], "claude-sonnet-4-6"); assert!(!captured.headers.to_ascii_lowercase().contains("anthropic-beta:")); @@ -4724,7 +5058,8 @@ mod tests { }); let stream = futures::stream::iter([Ok::<_, reqwest::Error>(bytes::Bytes::from(format!("data: {event}\n\n")))]); - let error = measure_first_stream_chunk(stream, std::time::Instant::now(), false, true).await.unwrap_err(); + let error = + measure_first_stream_chunk(stream, std::time::Instant::now(), false, true, false).await.unwrap_err(); assert!(error.contains("finishReason=MAX_TOKENS")); assert!(error.contains("thoughts=256")); @@ -4746,7 +5081,8 @@ mod tests { }); let stream = futures::stream::iter([Ok::<_, reqwest::Error>(bytes::Bytes::from(format!("data: {event}\n\n")))]); - let error = measure_first_stream_chunk(stream, std::time::Instant::now(), false, true).await.unwrap_err(); + let error = + measure_first_stream_chunk(stream, std::time::Instant::now(), false, true, false).await.unwrap_err(); assert!(error.contains("blockReason=SAFETY")); assert!(error.contains("HARM_CATEGORY_DANGEROUS_CONTENT:HIGH:blocked")); @@ -4774,13 +5110,15 @@ mod tests { #[tokio::test] async fn connection_probe_distinguishes_empty_body_from_non_sse_proxy_response() { let empty = futures::stream::empty::>(); - let empty_error = measure_first_stream_chunk(empty, std::time::Instant::now(), false, true).await.unwrap_err(); + let empty_error = + measure_first_stream_chunk(empty, std::time::Instant::now(), false, true, false).await.unwrap_err(); assert!(empty_error.contains("response body was empty")); assert!(empty_error.contains("endpoint or proxy")); assert_eq!(classify_error(&empty_error), "emptyResponse"); let proxy = futures::stream::iter([Ok::<_, reqwest::Error>(bytes::Bytes::from_static(b"proxy response"))]); - let proxy_error = measure_first_stream_chunk(proxy, std::time::Instant::now(), false, true).await.unwrap_err(); + let proxy_error = + measure_first_stream_chunk(proxy, std::time::Instant::now(), false, true, false).await.unwrap_err(); assert!(proxy_error.contains("14 bytes but no SSE data events")); assert!(proxy_error.contains("proxy streaming support")); assert_eq!(classify_error(&proxy_error), "emptyResponse"); @@ -4793,7 +5131,8 @@ mod tests { }); let stream = futures::stream::iter([Ok::<_, reqwest::Error>(bytes::Bytes::from(format!("data: {event}")))]); - let (_, text) = measure_first_stream_chunk(stream, std::time::Instant::now(), false, true).await.unwrap(); + let (_, text) = + measure_first_stream_chunk(stream, std::time::Instant::now(), false, true, false).await.unwrap(); assert_eq!(text, "OK"); } @@ -5290,9 +5629,19 @@ mod tests { assert_eq!(resolve_model_list_endpoint(&config).unwrap(), "https://api.minimax.io/v1/models"); assert!(provider_requires_api_key(&config.provider)); + let china = AiConfig { endpoint: "https://api.minimaxi.com/v1".to_string(), ..config.clone() }; + assert_eq!(resolve_endpoint(&china), "https://api.minimaxi.com/v1/chat/completions"); + assert_eq!(resolve_model_list_endpoint(&china).unwrap(), "https://api.minimaxi.com/v1/models"); + let no_key = AiConfig { api_key: String::new(), ..config.clone() }; assert_eq!(validate_config(&no_key).unwrap_err(), "API key is required"); + let responses = AiConfig { api_style: AiApiStyle::Responses, ..config.clone() }; + assert_eq!( + validate_config(&responses).unwrap_err(), + "MiniMax currently supports the Chat Completions API style in DBX; select Completions and retry" + ); + let provider_json = serde_json::to_string(&AiProvider::MiniMax).unwrap(); assert_eq!(provider_json, r#""minimax""#); assert!(matches!(serde_json::from_str::(&provider_json).unwrap(), AiProvider::MiniMax)); @@ -6357,6 +6706,353 @@ mod tests { assert!(body.get("reasoning_effort").is_none()); } + #[test] + fn minimax_chat_completion_decorator_uses_official_fields() { + let mut config = minimax_test_config("https://api.minimaxi.com/v1"); + config.enable_thinking = false; + let mut body = serde_json::json!({ "model": &config.model }); + + decorate_chat_completion_body(&mut body, &config, 1024); + + assert_eq!(body["reasoning_split"], true); + assert_eq!(body["max_completion_tokens"], 1024); + assert_eq!(body["thinking"]["type"], "disabled"); + assert!(body.get("max_tokens").is_none()); + assert!(body.get("extra_body").is_none()); + } + + #[tokio::test] + async fn minimax_completion_request_sends_official_chat_fields() { + let (capture_endpoint, server) = + spawn_json_capture_server("application/json", r#"{"choices":[{"message":{"content":"ok"}}]}"#).await; + let endpoint = capture_endpoint.trim_end_matches("/v1/messages").to_string(); + let mut request = minimax_test_request(endpoint); + request.config.enable_thinking = false; + let client = build_ai_http_client(&request.config, 10).unwrap(); + + assert_eq!(call_openai_compatible(&client, request).await.unwrap(), "ok"); + + let captured = server.await.unwrap(); + assert!(captured.headers.starts_with("POST /v1/chat/completions ")); + assert!(captured.headers.to_ascii_lowercase().contains("authorization: bearer minimax-test-key")); + assert_eq!(captured.body["reasoning_split"], true); + assert_eq!(captured.body["max_completion_tokens"], 64); + assert_eq!(captured.body["thinking"]["type"], "disabled"); + assert!(captured.body.get("max_tokens").is_none()); + } + + #[test] + fn minimax_cumulative_text_normalizer_emits_only_new_suffixes() { + let mut normalizer = MiniMaxTextAccumulator::default(); + + assert_eq!(normalizer.push("Hel"), Some("Hel".to_string())); + assert_eq!(normalizer.push("Hello"), Some("lo".to_string())); + assert_eq!(normalizer.push("Hello"), None); + assert_eq!(normalizer.push("Hell"), None); + assert_eq!(normalizer.push("Hello world"), Some(" world".to_string())); + assert_eq!(normalizer.push(""), None); + assert_eq!(normalizer.complete(), "Hello world"); + } + + #[test] + fn minimax_cumulative_text_normalizer_accepts_reset_or_incremental_chunks() { + let mut normalizer = MiniMaxTextAccumulator::default(); + + assert_eq!(normalizer.push("Hello"), Some("Hello".to_string())); + assert_eq!(normalizer.push(" world"), Some(" world".to_string())); + assert_eq!(normalizer.push("!"), Some("!".to_string())); + assert_eq!(normalizer.complete(), "Hello world!"); + } + + #[test] + fn minimax_stream_state_splits_cumulative_reasoning_and_content() { + let mut state = MiniMaxStreamState::default(); + let first_details = serde_json::json!([{ + "type": "reasoning.text", + "text": "Inspect" + }]); + let first = serde_json::json!({ + "choices": [{ + "delta": { + "content": "An", + "reasoning_details": first_details + } + }] + }); + assert_eq!( + state.process(&first), + MiniMaxStreamDelta { text: Some("An".to_string()), reasoning: Some("Inspect".to_string()) } + ); + + let latest_details = serde_json::json!([{ + "type": "reasoning.text", + "text": "Inspect schema" + }]); + let second = serde_json::json!({ + "choices": [{ + "delta": { + "content": "Answer", + "reasoning_details": latest_details + } + }] + }); + assert_eq!( + state.process(&second), + MiniMaxStreamDelta { text: Some("swer".to_string()), reasoning: Some(" schema".to_string()) } + ); + assert_eq!( + state.provider_payload(), + Some(serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: latest_details, + })) + ); + } + + #[test] + fn minimax_stream_state_reconstructs_incremental_reasoning_details_for_replay() { + let mut state = MiniMaxStreamState::default(); + let first = serde_json::json!({ + "choices": [{ + "delta": { + "reasoning_content": "Inspect", + "reasoning_details": [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "format": "MiniMax-response-v1", + "index": 0, + "text": "Inspect" + }] + } + }] + }); + let second = serde_json::json!({ + "choices": [{ + "delta": { + "reasoning_content": " schema", + "reasoning_details": [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "format": "MiniMax-response-v1", + "index": 0, + "text": " schema" + }] + } + }] + }); + + assert_eq!(state.process(&first).reasoning, Some("Inspect".to_string())); + assert_eq!(state.process(&second).reasoning, Some(" schema".to_string())); + assert_eq!( + state.provider_payload(), + Some(serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "format": "MiniMax-response-v1", + "index": 0, + "text": "Inspect schema" + }] + })) + ); + } + + #[test] + fn minimax_china_stream_keeps_incremental_fragments_that_share_a_prefix() { + let config = minimax_test_config("https://api.minimaxi.com/v1"); + let mut state = MiniMaxStreamState::new(minimax_stream_semantics(&config)); + let first = serde_json::json!({ + "choices": [{ + "delta": { + "content": "a", + "reasoning_details": [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0, + "text": "a" + }] + } + }] + }); + let second = serde_json::json!({ + "choices": [{ + "delta": { + "content": "and", + "reasoning_details": [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0, + "text": "and" + }] + } + }] + }); + + assert_eq!( + state.process(&first), + MiniMaxStreamDelta { text: Some("a".to_string()), reasoning: Some("a".to_string()) } + ); + assert_eq!( + state.process(&second), + MiniMaxStreamDelta { text: Some("and".to_string()), reasoning: Some("and".to_string()) } + ); + assert_eq!( + state.provider_payload(), + Some(serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0, + "text": "aand" + }] + })) + ); + } + + #[test] + fn minimax_stream_state_accepts_reasoning_content_without_details() { + let mut state = MiniMaxStreamState::default(); + let content_only = serde_json::json!({ + "choices": [{ + "delta": { + "reasoning_content": "Fallback reasoning" + } + }] + }); + let metadata_only_details = serde_json::json!({ + "choices": [{ + "delta": { + "reasoning_content": " continues", + "reasoning_details": [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0 + }] + } + }] + }); + + assert_eq!(state.process(&content_only).reasoning, Some("Fallback reasoning".to_string())); + assert_eq!(state.process(&metadata_only_details).reasoning, Some(" continues".to_string())); + assert_eq!( + state.provider_payload(), + Some(serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0 + }] + })) + ); + } + + #[test] + fn minimax_tool_history_replays_reasoning_details_once_for_parallel_calls() { + let details = serde_json::json!([{ + "type": "reasoning.text", + "text": "Need two lookups" + }]); + let payload = serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: details, + }); + let message = AiMessage { + role: "assistant".to_string(), + content: String::new(), + tool_call_id: None, + tool_calls: vec![ + ToolCallRef { + id: "call_1".to_string(), + name: "list_tables".to_string(), + arguments: serde_json::json!({"schema": "public"}), + provider_payload: Some(payload), + }, + ToolCallRef { + id: "call_2".to_string(), + name: "list_tables".to_string(), + arguments: serde_json::json!({"schema": "audit"}), + provider_payload: None, + }, + ], + }; + + let config = minimax_test_config("https://api.minimax.io/v1"); + let messages = build_openai_chat_messages(&config, "system", std::slice::from_ref(&message)); + assert_eq!(messages[1]["reasoning_details"], details); + assert_eq!(messages[1]["tool_calls"].as_array().unwrap().len(), 2); + assert_eq!(messages[1]["tool_calls"][0]["function"]["arguments"], r#"{"schema":"public"}"#); + + let mut compatible = config; + compatible.provider = AiProvider::OpenaiCompatible; + let generic = build_openai_chat_messages(&compatible, "system", &[message]); + assert!(generic[1].get("reasoning_details").is_none()); + } + + #[tokio::test] + async fn minimax_connection_probe_accepts_split_reasoning() { + let event = serde_json::json!({ + "choices": [{ + "delta": { + "reasoning_details": [{ + "type": "reasoning.text", + "text": "Ready" + }] + } + }] + }); + let stream = futures::stream::iter([Ok::<_, reqwest::Error>(bytes::Bytes::from(format!("data: {event}\n\n")))]); + + let (_, text) = + measure_first_stream_chunk(stream, std::time::Instant::now(), false, false, true).await.unwrap(); + + assert_eq!(text, "Ready"); + } + + #[tokio::test] + async fn minimax_tool_stream_preserves_reasoning_details_for_replay() { + let response = concat!( + "data: {\"choices\":[{\"delta\":{\"reasoning_details\":[{\"type\":\"reasoning.text\",\"id\":\"reasoning-text-1\",\"index\":0,\"text\":\"Inspect\"}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"reasoning_details\":[{\"type\":\"reasoning.text\",\"id\":\"reasoning-text-1\",\"index\":0,\"text\":\" schema\"}]}}]}\n\n", + "data: {\"choices\":[{\"delta\":{\"tool_calls\":[{\"index\":0,\"id\":\"call_1\",\"function\":{\"name\":\"list_tables\",\"arguments\":\"{\\\"schema\\\":\\\"public\\\"}\"}}]},\"finish_reason\":\"tool_calls\"}]}\n\n", + "data: [DONE]\n\n" + ); + let (capture_endpoint, server) = spawn_json_capture_server("text/event-stream", response).await; + let endpoint = capture_endpoint.trim_end_matches("/v1/messages").to_string(); + let request = minimax_test_request(endpoint); + let chunks = std::sync::Arc::new(std::sync::Mutex::new(Vec::new())); + let observed = chunks.clone(); + + let (calls, _) = + stream_with_tools(&request.config, &request, "minimax-tool-test", &[], &Notify::new(), move |chunk| { + observed.lock().unwrap().push(chunk) + }) + .await + .unwrap(); + + assert_eq!(calls.len(), 1); + assert_eq!(calls[0].id, "call_1"); + assert_eq!(calls[0].name, "list_tables"); + assert_eq!(calls[0].arguments, serde_json::json!({"schema": "public"})); + assert_eq!( + calls[0].provider_payload, + Some(serde_json::json!({ + MINIMAX_REASONING_DETAILS_PAYLOAD_KEY: [{ + "type": "reasoning.text", + "id": "reasoning-text-1", + "index": 0, + "text": "Inspect schema" + }] + })) + ); + assert_eq!( + chunks.lock().unwrap().iter().filter_map(|chunk| chunk.reasoning_delta.as_deref()).collect::(), + "Inspect schema" + ); + + let captured = server.await.unwrap(); + assert!(captured.headers.starts_with("POST /v1/chat/completions ")); + assert_eq!(captured.body["reasoning_split"], true); + assert_eq!(captured.body["max_completion_tokens"], 64); + } + #[test] fn runtime_provider_default_suppresses_legacy_thinking_toggle() { let mut config = AiConfig {