diff --git a/crates/tool_parser/src/parsers/minimax_m3.rs b/crates/tool_parser/src/parsers/minimax_m3.rs index c8f6392a3..0de2c378d 100644 --- a/crates/tool_parser/src/parsers/minimax_m3.rs +++ b/crates/tool_parser/src/parsers/minimax_m3.rs @@ -135,6 +135,30 @@ impl MinimaxM3Parser { .max() } + /// Whether a buffered tool-call wrapper can still become a valid invoke. + /// + /// Whitespace is allowed between the wrapper and the invoke marker. Once + /// the first non-whitespace bytes diverge from that marker, later input + /// cannot turn the candidate into a tool call. + fn could_start_invoke(buffer: &str) -> bool { + let Some(after_wrapper) = buffer.strip_prefix(TOOL_CALL_START) else { + return false; + }; + let candidate = after_wrapper.trim_start(); + if candidate.is_empty() || INVOKE_START.starts_with(candidate) { + return true; + } + + let Some(after_invoke) = candidate.strip_prefix(INVOKE_START) else { + return false; + }; + after_invoke.is_empty() + || after_invoke + .chars() + .next() + .is_some_and(|c| c.is_whitespace() || c == '>') + } + /// Decode common XML entities. fn decode_xml_entities(text: &str) -> String { text.replace("<", "<") @@ -484,6 +508,15 @@ impl ToolParser for MinimaxM3Parser { // Inside a tool call: wait for the complete end token before emitting. let Some(end_rel) = self.buffer.find(TOOL_CALL_END) else { + if !Self::could_start_invoke(&self.buffer) { + // Release the false wrapper, then resume the normal-text + // scan so a later marker (including a partial one) is still + // recognized rather than flushed as ordinary content. + normal_text.push_str(TOOL_CALL_START); + self.buffer.drain(..TOOL_CALL_START.len()); + self.in_tool_call = false; + continue; + } break; }; let block_end = end_rel + TOOL_CALL_END.len(); @@ -549,6 +582,14 @@ impl ToolParser for MinimaxM3Parser { helpers::get_unstreamed_args(&self.prev_tool_call_arr, &self.streamed_args_for_tool) } + fn take_unstreamed_normal_text(&mut self) -> String { + // Completed blocks are removed from `buffer`, so anything left here is + // an independent, incomplete candidate and must be returned verbatim. + // Leave the parser ready to process ordinary text if it is reused. + self.in_tool_call = false; + std::mem::take(&mut self.buffer) + } + fn reset(&mut self) { self.buffer.clear(); self.prev_tool_call_arr.clear(); diff --git a/crates/tool_parser/tests/tool_parser_minimax_m3.rs b/crates/tool_parser/tests/tool_parser_minimax_m3.rs index fea65a284..318894efa 100644 --- a/crates/tool_parser/tests/tool_parser_minimax_m3.rs +++ b/crates/tool_parser/tests/tool_parser_minimax_m3.rs @@ -420,6 +420,112 @@ async fn test_m3_streaming_no_markers_passthrough() { assert_eq!(normal, "Hello, world!"); } +#[tokio::test] +async fn test_m3_streaming_wrapper_and_invoke_across_every_chunk_boundary() { + let tools = create_test_tools(); + let full = tool_block(&[("get_weather", element("city", "Seattle"))]); + let invoke_header = "name=\"get_weather\">"; + let prefix_end = full.find(invoke_header).unwrap() + invoke_header.len(); + + for split in 1..prefix_end { + let mut parser = MinimaxM3Parser::new(); + + let first = parser + .parse_incremental(&full[..split], &tools) + .await + .unwrap(); + assert!(first.normal_text.is_empty(), "split {split}"); + assert!(first.calls.is_empty(), "split {split}"); + + let second = parser + .parse_incremental(&full[split..], &tools) + .await + .unwrap(); + assert!(second.normal_text.is_empty(), "split {split}"); + assert_eq!( + second.calls.iter().find_map(|call| call.name.as_deref()), + Some("get_weather"), + "split {split}" + ); + } +} + +#[tokio::test] +async fn test_m3_streaming_false_invoke_prefix_recovers_at_every_divergence() { + let tools = create_test_tools(); + let wrapper = format!("{NS}"); + let possible_invoke = format!("\n\t{NS}"); + let candidates = [ + wrapper[..wrapper.len() - 1].to_string(), + format!("{wrapper}\n {NS}"), + ]; + + for candidate in candidates { + let mut parser = MinimaxM3Parser::new(); + let result = parser.parse_incremental(&candidate, &tools).await.unwrap(); + assert!(result.normal_text.is_empty(), "candidate {candidate:?}"); + assert!(result.calls.is_empty(), "candidate {candidate:?}"); + + assert_eq!(parser.take_unstreamed_normal_text(), candidate); + assert_eq!(parser.take_unstreamed_normal_text(), ""); + + let next = parser + .parse_incremental("ordinary text", &tools) + .await + .unwrap(); + assert_eq!(next.normal_text, "ordinary text"); + assert!(next.calls.is_empty()); + } +} + +#[tokio::test] +async fn test_m3_streaming_eof_returns_new_candidate_after_completed_call() { + let tools = create_test_tools(); + let mut parser = MinimaxM3Parser::new(); + let call = tool_block(&[("get_weather", element("city", "Seattle"))]); + + let complete = parser.parse_incremental(&call, &tools).await.unwrap(); + assert_eq!( + complete.calls.iter().find_map(|item| item.name.as_deref()), + Some("get_weather") + ); + + let incomplete = format!("{NS}\n{NS}"); + let pending = parser.parse_incremental(&incomplete, &tools).await.unwrap(); + assert!(pending.normal_text.is_empty()); + assert!(pending.calls.is_empty()); + assert_eq!(parser.take_unstreamed_normal_text(), incomplete); +} + #[tokio::test] async fn test_m3_reset_between_requests() { let mut parser = MinimaxM3Parser::new();