launcher/docker-images/mimo26-spark/patches/sglang/40633.diff

diff --git a/python/sglang/srt/function_call/mimo_detector.py b/python/sglang/srt/function_call/mimo_detector.py
index 1af745c82d88..6a273f6a4bcd 100644
--- a/python/sglang/srt/function_call/mimo_detector.py
+++ b/python/sglang/srt/function_call/mimo_detector.py
@@ -12,7 +12,6 @@
 # limitations under the License.
 # ==============================================================================
 
-import html
 import json
 import logging
 import re
@@ -24,6 +23,7 @@
 from sglang.srt.function_call.core_types import StreamingParseResult, _GetInfoFunc
 from sglang.srt.function_call.utils import (
     get_schema_properties,
+    infer_type_from_json_schema,
     safe_literal_eval,
 )
 
@@ -36,7 +36,7 @@ def _get_param_type(func_name: str, param_name: str, tools: List[Tool]) -> str:
         if tool.function.name == func_name:
             props = get_schema_properties(tool.function.parameters)
             if param_name in props:
-                return props[param_name].get("type", "string")
+                return infer_type_from_json_schema(props[param_name]) or "string"
     return "string"
 
 
@@ -47,16 +47,12 @@ def _convert_param_value(
     Convert parameter value based on its type in the schema.
     Adapted from vllm-project/vllm (vllm/entrypoints/openai/tool_parsers/qwen3coder_tool_parser.py)
     """
-    param_value = html.unescape(param_value)
-
-    # Handle null value for any type
-    if param_value.lower() == "null":
-        return None
-
     param_type = _get_param_type(func_name, param_name, tools)
 
     if param_type in ["string", "str", "text", "varchar", "char", "enum"]:
         return param_value
+    elif param_value == "null":
+        return None
     elif (
         param_type.startswith("int")
         or param_type.startswith("integer")
@@ -278,7 +274,88 @@ def _parse_tool_call(
         return {"name": func_name, "parameters": params}
 
     def supports_structural_tag(self) -> bool:
-        return False
+        return True
 
     def structure_info(self) -> _GetInfoFunc:
         raise NotImplementedError
+
+    def get_structural_tag_name(self) -> str:
+        return "mimo"
+
+    def get_structural_tag(
+        self,
+        tools=None,
+        tool_choice="auto",
+        thinking_mode=False,
+        parallel_tool_calls=True,
+    ):
+        try:
+            return super().get_structural_tag(
+                tools, tool_choice, thinking_mode, parallel_tool_calls
+            )
+        except ValueError as error:
+            if not str(error).startswith("Unknown format type: mimo,"):
+                raise
+
+        # XGrammar 0.2.7 has qwen_xml, but predates the compact MiMo builtin.
+        from xgrammar import normalize_tool_choice
+        from xgrammar.structural_tag import (
+            AnyTextFormat,
+            JSONSchemaFormat,
+            SequenceFormat,
+            StructuralTag,
+            TagFormat,
+            TriggeredTagsFormat,
+        )
+
+        from sglang.srt.entrypoints.openai.protocol import ToolChoice
+
+        functions, builtins, choice = normalize_tool_choice(
+            [tool.model_dump() for tool in tools or []],
+            tool_choice.model_dump()
+            if isinstance(tool_choice, ToolChoice)
+            else tool_choice,
+        )
+        if builtins:
+            raise ValueError("MiMo does not support builtin tools.")
+        tags = [
+            TagFormat(
+                begin=f"<tool_call><function={tool.function.name}>",
+                content=JSONSchemaFormat(
+                    json_schema=(
+                        tool.function.parameters
+                        if tool.function.strict is not False
+                        and tool.function.parameters
+                        else True
+                    ),
+                    style="qwen_xml",
+                ),
+                end="</function></tool_call>",
+            )
+            for tool in functions
+        ]
+        excludes = ["<think>", "</think>", "</tool_call>", "<function="]
+        if choice == "forced":
+            suffix = tags[0]
+        elif tags:
+            suffix = TriggeredTagsFormat(
+                triggers=["<tool_call>"],
+                tags=tags,
+                excludes=excludes,
+                at_least_one=choice == "required",
+                stop_after_first=not parallel_tool_calls,
+            )
+        else:
+            suffix = AnyTextFormat(excludes=["<tool_call>", *excludes])
+        if thinking_mode:
+            suffix = SequenceFormat(
+                elements=[
+                    TagFormat(
+                        begin="<think>",
+                        content=AnyTextFormat(excludes=["<tool_call>", *excludes]),
+                        end="</think>",
+                    ),
+                    suffix,
+                ]
+            )
+        return StructuralTag(format=suffix)
diff --git a/test/registered/unit/function_call/test_function_call_parser.py b/test/registered/unit/function_call/test_function_call_parser.py
index 359ae726dbfc..20aa22a5f5f9 100644
--- a/test/registered/unit/function_call/test_function_call_parser.py
+++ b/test/registered/unit/function_call/test_function_call_parser.py
@@ -33,6 +33,7 @@
 from sglang.srt.function_call.lfm2_detector import Lfm2Detector
 from sglang.srt.function_call.ling3_detector import Ling3Detector
 from sglang.srt.function_call.llama32_detector import Llama32Detector
+from sglang.srt.function_call.mimo_detector import MiMoDetector
 from sglang.srt.function_call.mistral_detector import MistralDetector
 from sglang.srt.function_call.parser_names import TOOL_CALL_PARSER_NAMES
 from sglang.srt.function_call.pythonic_detector import PythonicDetector
@@ -6024,5 +6025,137 @@ def test_matches_registry(self):
         )
 
 
+class TestMiMoStructuralTag(unittest.TestCase):
+    def setUp(self):
+        self.tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="run",
+                    strict=True,
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "text": {"type": "string"},
+                            "n": {"type": ["integer", "null"]},
+                            "data": {"type": "object"},
+                        },
+                        "required": ["text", "n", "data"],
+                        "additionalProperties": False,
+                    },
+                ),
+            )
+        ]
+        self.call = (
+            "<tool_call><function=run><parameter=text>\n北京 &amp;\n</parameter>"
+            '<parameter=n>2</parameter><parameter=data>{"items": [true, null]}</parameter></function></tool_call>'
+        )
+
+    def test_compact_xml_schema_constraints(self):
+        from xgrammar.testing import _is_grammar_accept_string
+
+        for policy in [
+            "auto",
+            "required",
+            ToolChoice(type="function", function=ToolChoiceFuncName(name="run")),
+        ]:
+            for thinking in [False, True]:
+                with self.subTest(policy=policy, thinking=thinking):
+                    detector = MiMoDetector()
+                    self.assertTrue(detector.supports_structural_tag())
+                    tag = detector.get_structural_tag(
+                        self.tools, policy, thinking_mode=thinking
+                    )
+                    grammar = xgr.Grammar.from_structural_tag(tag)
+                    prefix = "<think>plan</think>" if thinking else ""
+                    self.assertTrue(
+                        _is_grammar_accept_string(grammar, prefix + self.call)
+                    )
+                    self.assertEqual(
+                        _is_grammar_accept_string(grammar, prefix + "answer"),
+                        policy == "auto",
+                    )
+                    for invalid in [
+                        self.call.replace("function=run", "function=unknown"),
+                        self.call.replace("<parameter=n>2</parameter>", ""),
+                        self.call.replace("<parameter=n>2", "<parameter=n>bad"),
+                        self.call.replace("<parameter=n>", "<parameter=extra>"),
+                        self.call.replace(
+                            "</function>", "<parameter=n>3</parameter></function>"
+                        ),
+                    ]:
+                        self.assertFalse(
+                            _is_grammar_accept_string(grammar, prefix + invalid)
+                        )
+                    self.assertEqual(
+                        _is_grammar_accept_string(grammar, prefix + self.call * 2),
+                        isinstance(policy, str),
+                    )
+                    single = xgr.Grammar.from_structural_tag(
+                        detector.get_structural_tag(
+                            self.tools,
+                            policy,
+                            thinking_mode=thinking,
+                            parallel_tool_calls=False,
+                        )
+                    )
+                    self.assertTrue(
+                        _is_grammar_accept_string(single, prefix + self.call)
+                    )
+                    self.assertFalse(
+                        _is_grammar_accept_string(single, prefix + self.call * 2)
+                    )
+
+    def test_non_strict_keeps_native_parameter_format(self):
+        from xgrammar.testing import _is_grammar_accept_string
+
+        self.tools[0].function.strict = False
+        grammar = xgr.Grammar.from_structural_tag(
+            MiMoDetector().get_structural_tag(self.tools, "required")
+        )
+        self.assertTrue(
+            _is_grammar_accept_string(
+                grammar,
+                "<tool_call><function=run><parameter=extra>anything</parameter></function></tool_call>",
+            )
+        )
+
+    def test_parameters_round_trip_without_html_or_string_coercion(self):
+        for text in ["\n北京 &amp;\n", "null", "42", ""]:
+            for number in ["2", "null"]:
+                with self.subTest(text=text, number=number):
+                    call = self.call.replace("\n北京 &amp;\n", text).replace(
+                        "<parameter=n>2", "<parameter=n>" + number
+                    )
+                    result = MiMoDetector().detect_and_parse(call, self.tools)
+                    self.assertEqual(
+                        json.loads(result.calls[0].parameters),
+                        {
+                            "text": text,
+                            "n": json.loads(number),
+                            "data": {"items": [True, None]},
+                        },
+                    )
+
+    def test_streamed_compact_calls_preserve_arguments(self):
+        # MiMo's tool-call markers are atomic tokenizer tokens; parameter tags are not.
+        chunks = [
+            "<tool_call>",
+            *self.call[len("<tool_call>") : -len("</tool_call>")],
+            "</tool_call>",
+        ]
+        detector = MiMoDetector()
+        calls = []
+        for chunk in chunks * 2:
+            calls.extend(detector.parse_streaming_increment(chunk, self.tools).calls)
+        self.assertEqual(len(calls), 2)
+        self.assertEqual([call.tool_index for call in calls], [0, 1])
+        for call in calls:
+            self.assertEqual(
+                json.loads(call.parameters),
+                {"text": "\n北京 &amp;\n", "n": 2, "data": {"items": [True, None]}},
+            )
+
+
 if __name__ == "__main__":
     unittest.main()