diff --git a/python/sglang/srt/function_call/glm47_moe_detector.py b/python/sglang/srt/function_call/glm47_moe_detector.py
index 726737c57..0cf3a1b62 100644
--- a/python/sglang/srt/function_call/glm47_moe_detector.py
+++ b/python/sglang/srt/function_call/glm47_moe_detector.py
@@ -40,15 +40,27 @@ def get_argument_type(
         The type string (e.g., 'string', 'number', 'object') or None if not found
     """
     name2tool = {tool.function.name: tool for tool in defined_tools}
-    if func_name not in name2tool:
+
+    # Check if function exists
+    tool = name2tool.get(func_name)
+    if not tool:
         return None
-    tool = name2tool[func_name]
-    properties = (tool.function.parameters or {}).get("properties", {})
+
+    # Get parameters safely using getattr
+    params = getattr(tool.function, "parameters", None)
+    if not isinstance(params, dict):
+        return None
+
+    # Navigate to the type using dict.get() for safe access
+    properties = params.get("properties")
     if not isinstance(properties, dict):
-        properties = {}
-    if arg_key not in properties:
         return None
-    return properties[arg_key].get("type", None)
+
+    arg_spec = properties.get(arg_key)
+    if isinstance(arg_spec, dict):
+        return arg_spec.get("type")
+
+    return None
 
 
 def _convert_to_number(value: str) -> Any:
@@ -143,6 +155,10 @@ class Glm47MoeDetector(BaseFormatDetector):
         self.current_tool_id = -1
         self.current_tool_name_sent = False
         self._streamed_raw_length = 0
+        self._tool_call_completed = False  # Track if tool call has been completed
+        self._sent_empty_object = (
+            False  # Track if empty object has been sent for no-arg functions
+        )
         self._reset_streaming_state()
 
     def _reset_streaming_state(self) -> None:
@@ -156,6 +172,8 @@ class Glm47MoeDetector(BaseFormatDetector):
         self._cached_value_type: Optional[str] = (
             None  # Cache the value type for consistency
         )
+        self._tool_call_completed = False  # Reset tool call completion status
+        self._sent_empty_object = False  # Reset empty object sent status
 
     def has_tool_call(self, text: str) -> bool:
         """Check if the text contains a glm-4.5 / glm-4.6 format tool call."""
@@ -169,18 +187,38 @@ class Glm47MoeDetector(BaseFormatDetector):
         :param tools: List of available tools.
         :return: ParseResult indicating success or failure, consumed text, leftover text, and parsed calls.
         """
-        idx = text.find(self.bot_token)
-        normal_text = text[:idx].strip() if idx != -1 else text
         if self.bot_token not in text:
-            return StreamingParseResult(normal_text=normal_text, calls=[])
+            return StreamingParseResult(normal_text=text, calls=[])
+
+        # Extract all normal text (before, between, and after tool calls)
+        normal_text_parts = []
+        last_end = 0
+
+        # Find all tool call matches
+        for match in re.finditer(self.func_call_regex, text, re.DOTALL):
+            # Add text before this tool call
+            if match.start() > last_end:
+                normal_text_parts.append(text[last_end : match.start()])
+            last_end = match.end()
+
+        # Add any remaining text after the last tool call
+        if last_end < len(text):
+            normal_text_parts.append(text[last_end:])
+
+        # Combine all normal text parts
+        normal_text = "".join(normal_text_parts).strip()
+
+        # Parse tool calls
         match_result_list = re.findall(self.func_call_regex, text, re.DOTALL)
         calls = []
         try:
             for match_result in match_result_list:
                 # Get function name
                 func_detail = self.func_detail_regex.search(match_result)
-                func_name = func_detail.group(1)
-                func_args = func_detail.group(2)
+                if func_detail is None:
+                    continue
+                func_name = func_detail.group(1) if func_detail.group(1) else ""
+                func_args = func_detail.group(2) if func_detail.group(2) else ""
                 arguments = {}
                 if func_args:
                     pairs = self.func_arg_regex.findall(func_args)
@@ -212,7 +250,11 @@ class Glm47MoeDetector(BaseFormatDetector):
             return arg_type
 
         # Auto-detect type from value (best effort)
-        first_chars = self._current_value.strip()[:10] if self._current_value else ""
+        first_chars = (
+            self._current_value.strip()[:10]
+            if self._current_value and self._current_value.strip()
+            else ""
+        )
         if first_chars:
             first_char = first_chars[0]
             if first_char.isdigit() or first_char in ["-", "."]:
@@ -237,14 +279,14 @@ class Glm47MoeDetector(BaseFormatDetector):
             return json.dumps(value, ensure_ascii=False)
         elif value_type == "number":
             try:
-                num = _convert_to_number(value.strip())
+                num = _convert_to_number(value.strip() if value else "")
                 return str(num)
             except (ValueError, AttributeError):
                 # Fallback to string if not a valid number
                 logger.warning(
                     f"Failed to parse '{value}' as number, treating as string"
                 )
-                return json.dumps(str(value), ensure_ascii=False)
+                return json.dumps(str(value) if value else "", ensure_ascii=False)
         else:
             # For object/array types, return as-is (should already be valid JSON)
             return value
@@ -369,6 +411,179 @@ class Glm47MoeDetector(BaseFormatDetector):
 
         return json_output
 
+    def _extract_match_groups(self, match: re.Match) -> tuple[str, str, str]:
+        """Extract function name, arguments and end marker from regex match.
+
+        Args:
+            match: Regex match object
+
+        Returns:
+            (func_name, func_args_raw, is_tool_end)
+        """
+        func_name = match.group(1).strip()
+        func_args_raw = match.group(2).strip() if match.group(2) else ""
+        is_tool_end = match.group(3) or ""
+        return func_name, func_args_raw, is_tool_end
+
+    def _send_tool_name_if_needed(
+        self, func_name: str, has_arg_key: bool, is_tool_end: str
+    ) -> Optional[ToolCallItem]:
+        """Send tool name if needed.
+
+        Args:
+            func_name: Function name
+            has_arg_key: Whether current text contains <arg_key
+            is_tool_end: Whether end marker is encountered
+
+        Returns:
+            Tool call item or None
+        """
+        if self.current_tool_name_sent:
+            return None
+
+        # Function name completeness check
+        is_func_name_complete = has_arg_key or is_tool_end == self.eot_token
+
+        if not is_func_name_complete:
+            return None
+
+        if not func_name:
+            logger.warning("Empty function name detected, skipping tool call")
+            return None
+
+        # Send tool name
+        self.current_tool_name_sent = True
+        self._streamed_raw_length = 0
+        self._reset_streaming_state()
+
+        # Record tool info
+        self.prev_tool_call_arr[self.current_tool_id] = {
+            "name": func_name,
+            "arguments": {},
+        }
+
+        return ToolCallItem(
+            tool_index=self.current_tool_id,
+            name=func_name,
+            parameters="",
+        )
+
+    def _process_arguments_streaming(
+        self, func_name: str, func_args_raw: str, tools: List[Tool]
+    ) -> Optional[ToolCallItem]:
+        """Process streaming arguments.
+
+        Args:
+            func_name: Function name
+            func_args_raw: Raw argument string
+            tools: List of available tools
+
+        Returns:
+            Tool call item with parameter updates or None
+        """
+        current_raw_length = len(func_args_raw)
+
+        if current_raw_length <= self._streamed_raw_length:
+            return None
+
+        # Get new raw XML content
+        raw_increment = func_args_raw[self._streamed_raw_length :]
+
+        # Convert XML to JSON using state machine
+        json_increment = self._process_xml_to_json_streaming(
+            raw_increment, func_name, tools
+        )
+
+        # CRITICAL: Update streamed length BEFORE early return
+        # Even if json_increment is empty, the input has been consumed by the state machine
+        self._streamed_raw_length = current_raw_length
+
+        if not json_increment:
+            return None
+
+        # Update state
+        self._last_arguments += json_increment
+        self.streamed_args_for_tool[self.current_tool_id] += json_increment
+
+        return ToolCallItem(
+            tool_index=self.current_tool_id,
+            name=None,
+            parameters=json_increment,
+        )
+
+    def _finalize_tool_call(
+        self,
+        func_name: str,
+        func_args_raw: str,
+        tools: List[Tool],
+        match_end_pos: int,
+        current_text: str,
+    ) -> List[ToolCallItem]:
+        """Complete tool call processing.
+
+        Args:
+            func_name: Function name
+            func_args_raw: Raw argument string
+            tools: List of available tools
+            match_end_pos: Match end position
+            current_text: Current text
+
+        Returns:
+            List of tool call items to add
+        """
+        calls = []
+
+        # Handle no-arg function or need to close braces
+        if self._is_first_param and not self._sent_empty_object:
+            # No-arg function
+            calls.append(
+                ToolCallItem(
+                    tool_index=self.current_tool_id,
+                    name=None,
+                    parameters="{}",
+                )
+            )
+            self._last_arguments += "{}"
+            self.streamed_args_for_tool[self.current_tool_id] += "{}"
+            self._sent_empty_object = True
+        elif not self._last_arguments.endswith("}") and not self._sent_empty_object:
+            # Need to close brace
+            calls.append(
+                ToolCallItem(
+                    tool_index=self.current_tool_id,
+                    name=None,
+                    parameters="}",
+                )
+            )
+            self._last_arguments += "}"
+            self.streamed_args_for_tool[self.current_tool_id] += "}"
+            self._sent_empty_object = True
+
+        # Parse final arguments
+        if func_args_raw:
+            try:
+                pairs = self.func_arg_regex.findall(func_args_raw)
+                if pairs:
+                    arguments = self._parse_argument_pairs(pairs, func_name, tools)
+                    self.prev_tool_call_arr[self.current_tool_id][
+                        "arguments"
+                    ] = arguments
+            except Exception as e:
+                logger.debug(f"Failed to parse arguments: {e}", exc_info=True)
+
+        # Clean buffer
+        self._buffer = current_text[match_end_pos:]
+
+        # Reset state for next tool call
+        self._tool_call_completed = True
+        self.current_tool_id += 1
+        self._last_arguments = ""
+        self.current_tool_name_sent = False
+        self._streamed_raw_length = 0
+        self._reset_streaming_state()
+
+        return calls
+
     def parse_streaming_increment(
         self, new_text: str, tools: List[Tool]
     ) -> StreamingParseResult:
@@ -405,140 +620,96 @@ class Glm47MoeDetector(BaseFormatDetector):
                 # Could be start of tool call, keep buffering
                 return StreamingParseResult(normal_text="", calls=[])
 
+        # Extract any text before the first bot_token and return it as normal_text
+        normal_text = ""
+        first_bot_token_idx = current_text.find(self.bot_token)
+        if first_bot_token_idx > 0:
+            normal_text = current_text[:first_bot_token_idx]
+            current_text = current_text[first_bot_token_idx:]
+            # Update buffer to only include from the bot token onwards
+            self._buffer = current_text
+
         if not hasattr(self, "_tool_indices"):
             self._tool_indices = self._get_tool_indices(tools)
 
         calls: list[ToolCallItem] = []
         try:
             # Try to match a partial or complete tool call
+            # Use a single flexible regex pattern that handles all cases
             partial_match = re.search(
-                pattern=r"<tool_call>(.*?)(<arg_key>.*?)?(</tool_call>|$)",
-                string=current_text,
-                flags=re.DOTALL,
+                r"<tool_call>(.*?)(?:(<arg_key.*?))?(?:(</tool_call>)|$)",
+                current_text,
+                re.DOTALL,
             )
-            if partial_match:
-                func_name = partial_match.group(1).strip()
-                func_args_raw = partial_match.group(2).strip()
-                is_tool_end = partial_match.group(3)
 
-                # Initialize state if this is the first tool call
-                if self.current_tool_id == -1:
-                    self.current_tool_id = 0
-                    self.prev_tool_call_arr = []
-                    self.streamed_args_for_tool = [""]
-                    self._streamed_raw_length = 0
-                    self.current_tool_name_sent = False
-                    self._reset_streaming_state()
+            if not partial_match:
+                return StreamingParseResult(normal_text=normal_text, calls=[])
 
-                # Ensure we have enough entries in our tracking arrays
-                while len(self.prev_tool_call_arr) <= self.current_tool_id:
-                    self.prev_tool_call_arr.append({})
-                while len(self.streamed_args_for_tool) <= self.current_tool_id:
-                    self.streamed_args_for_tool.append("")
+            # Extract match groups using helper method
+            func_name, func_args_raw, is_tool_end = self._extract_match_groups(
+                partial_match
+            )
 
-                # Send tool name first if not sent yet
-                if not self.current_tool_name_sent:
-                    assert func_name, "func_name should not be empty"
-                    calls.append(
-                        ToolCallItem(
-                            tool_index=self.current_tool_id,
-                            name=func_name,
-                            parameters="",
-                        )
+            # Initialize tool call state if needed (keeping existing logic)
+            if self.current_tool_id == -1:
+                self.current_tool_id = 0
+                self.prev_tool_call_arr = []
+                self.streamed_args_for_tool = [""]
+                self._streamed_raw_length = 0
+                self.current_tool_name_sent = False  # Reset for new tool call
+                self._reset_streaming_state()
+            # Check if this is a continuation of an existing tool call or a new one
+            elif not self.current_tool_name_sent:
+                # Only increment tool_id if we're truly starting a NEW tool call
+                # Don't increment if this is just the first time we're processing
+                # a tool call that was received in the buffer
+                # The key insight: only increment when we've COMPLETED a previous tool call
+                # and now see another bot_token in new_text
+                pass  # Remove the problematic auto-increment logic
+
+            # Ensure tracking arrays are large enough (keeping existing logic)
+            while len(self.prev_tool_call_arr) <= self.current_tool_id:
+                self.prev_tool_call_arr.append({})
+            while len(self.streamed_args_for_tool) <= self.current_tool_id:
+                self.streamed_args_for_tool.append("")
+
+            # Determine if function name is complete by checking for <arg_key> in the full text
+            # This is important for streaming scenarios where args come in later chunks
+            has_arg_key = "<arg_key" in current_text
+
+            # Send tool name if needed
+            tool_name_item = self._send_tool_name_if_needed(
+                func_name, has_arg_key, is_tool_end
+            )
+            if tool_name_item:
+                calls.append(tool_name_item)
+
+            # Process streaming arguments if tool name has been sent
+            if self.current_tool_name_sent:
+                arg_item = self._process_arguments_streaming(
+                    func_name, func_args_raw, tools
+                )
+                if arg_item:
+                    calls.append(arg_item)
+
+                # Finalize tool call if end token is encountered
+                if is_tool_end == self.eot_token and not self._tool_call_completed:
+                    finalize_calls = self._finalize_tool_call(
+                        func_name,
+                        func_args_raw,
+                        tools,
+                        partial_match.end(),
+                        current_text,
                     )
-                    self.current_tool_name_sent = True
-                    self._streamed_raw_length = 0
-                    self._reset_streaming_state()
-                    # Store the tool call info
-                    self.prev_tool_call_arr[self.current_tool_id] = {
-                        "name": func_name,
-                        "arguments": {},
-                    }
-                else:
-                    # Process XML to JSON streaming
-                    current_raw_length = len(func_args_raw)
-
-                    if current_raw_length > self._streamed_raw_length:
-                        # Get the new raw XML content
-                        raw_increment = func_args_raw[self._streamed_raw_length :]
-
-                        # Convert XML increment to JSON increment using state machine
-                        json_increment = self._process_xml_to_json_streaming(
-                            raw_increment, func_name, tools
-                        )
-
-                        if json_increment:
-                            calls.append(
-                                ToolCallItem(
-                                    tool_index=self.current_tool_id,
-                                    name=None,
-                                    parameters=json_increment,
-                                )
-                            )
-                            self._last_arguments += json_increment
-                            self.streamed_args_for_tool[
-                                self.current_tool_id
-                            ] += json_increment
-
-                        # Update the streamed length
-                        self._streamed_raw_length = current_raw_length
-
-                    if is_tool_end == self.eot_token:
-                        if self._is_first_param:
-                            empty_object = "{}"
-                            calls.append(
-                                ToolCallItem(
-                                    tool_index=self.current_tool_id,
-                                    name=None,
-                                    parameters=empty_object,
-                                )
-                            )
-                            self._last_arguments += empty_object
-                        elif not self._last_arguments.endswith("}"):
-                            closing_brace = "}"
-                            calls.append(
-                                ToolCallItem(
-                                    tool_index=self.current_tool_id,
-                                    name=None,
-                                    parameters=closing_brace,
-                                )
-                            )
-                            self._last_arguments += closing_brace
-                            self.streamed_args_for_tool[
-                                self.current_tool_id
-                            ] += closing_brace
-
-                        try:
-                            pairs = self.func_arg_regex.findall(func_args_raw)
-                            if pairs:
-                                arguments = self._parse_argument_pairs(
-                                    pairs, func_name, tools
-                                )
-                                self.prev_tool_call_arr[self.current_tool_id][
-                                    "arguments"
-                                ] = arguments
-                        except Exception as e:
-                            logger.debug(
-                                f"Failed to parse arguments: {e}", exc_info=True
-                            )
-
-                        # Remove the completed tool call from buffer
-                        self._buffer = current_text[partial_match.end(3) :]
-
-                        result = StreamingParseResult(normal_text="", calls=calls)
-                        self.current_tool_id += 1
-                        self._last_arguments = ""
-                        self.current_tool_name_sent = False
-                        self._streamed_raw_length = 0
-                        self._reset_streaming_state()
-                        return result
-
-            return StreamingParseResult(normal_text="", calls=calls)
+                    calls.extend(finalize_calls)
+                    return StreamingParseResult(normal_text=normal_text, calls=calls)
 
         except Exception as e:
             logger.error(f"Error in parse_streaming_increment: {e}", exc_info=True)
             return StreamingParseResult(normal_text=current_text)
 
+        return StreamingParseResult(normal_text=normal_text, calls=calls)
+
     def _parse_argument_pairs(
         self, pairs: List[Tuple[str, str]], func_name: str, tools: List[Tool]
     ) -> Dict[str, Any]:
diff --git a/python/sglang/srt/function_call/glm4_moe_detector.py b/python/sglang/srt/function_call/glm4_moe_detector.py
index cf2201a5a..24b376ad9 100644
--- a/python/sglang/srt/function_call/glm4_moe_detector.py
+++ b/python/sglang/srt/function_call/glm4_moe_detector.py
@@ -189,8 +189,10 @@ class Glm4MoeDetector(BaseFormatDetector):
             for match_result in match_result_list:
                 # Get function name
                 func_detail = self.func_detail_regex.search(match_result)
-                func_name = func_detail.group(1)
-                func_args = func_detail.group(2)
+                if func_detail is None:
+                    continue
+                func_name = func_detail.group(1) if func_detail.group(1) else ""
+                func_args = func_detail.group(2) if func_detail.group(2) else ""
                 pairs = self.func_arg_regex.findall(func_args)
 
                 # Parse arguments using shared method
@@ -426,10 +428,19 @@ class Glm4MoeDetector(BaseFormatDetector):
                 flags=re.DOTALL,
             )
             if partial_match:
-                func_name = partial_match.group(1).strip()
-                func_args_raw = partial_match.group(2).strip()
+                func_name_raw = partial_match.group(1)
+                func_args_raw = partial_match.group(2)
                 is_tool_end = partial_match.group(3)
 
+                # Only proceed if we have a non-empty function name
+                if func_name_raw is None or not func_name_raw.strip():
+                    # If we only have the start token without a function name,
+                    # continue buffering until we get more content
+                    return StreamingParseResult(normal_text="", calls=[])
+
+                func_name = func_name_raw.strip()
+                func_args_raw = func_args_raw.strip() if func_args_raw else ""
+
                 # Initialize state if this is the first tool call
                 if self.current_tool_id == -1:
                     self.current_tool_id = 0
@@ -447,7 +458,6 @@ class Glm4MoeDetector(BaseFormatDetector):
 
                 # Send tool name first if not sent yet
                 if not self.current_tool_name_sent:
-                    assert func_name, "func_name should not be empty"
                     calls.append(
                         ToolCallItem(
                             tool_index=self.current_tool_id,
@@ -476,6 +486,10 @@ class Glm4MoeDetector(BaseFormatDetector):
                             raw_increment, func_name, tools
                         )
 
+                        # CRITICAL: Update streamed length BEFORE checking json_increment
+                        # Even if json_increment is empty, the input has been consumed by the state machine
+                        self._streamed_raw_length = current_raw_length
+
                         if json_increment:
                             calls.append(
                                 ToolCallItem(
@@ -489,9 +503,6 @@ class Glm4MoeDetector(BaseFormatDetector):
                                 self.current_tool_id
                             ] += json_increment
 
-                        # Update the streamed length
-                        self._streamed_raw_length = current_raw_length
-
                     if is_tool_end == self.eot_token:
                         if self._is_first_param:
                             empty_object = "{}"
diff --git a/test/registered/function_call/test_function_call_parser.py b/test/registered/function_call/test_function_call_parser.py
index c6b1cb352..d23658646 100644
--- a/test/registered/function_call/test_function_call_parser.py
+++ b/test/registered/function_call/test_function_call_parser.py
@@ -2257,6 +2257,21 @@ class TestGlm4MoeDetector(unittest.TestCase):
         )
         check_single_todos(result, expected_output)
 
+    def test_empty_function_name_handling(self):
+        """Test that empty function name is handled gracefully without assertion error."""
+        # This test simulates the issue where the model outputs only the start token without a function name
+        chunks = [
+            "<tool_call>",  # Start token only, no function name yet
+            "\n",  # More content without function name
+        ]
+
+        for chunk in chunks:
+            # Should not raise AssertionError: func_name should not be empty
+            result = self.detector.parse_streaming_increment(chunk, self.tools)
+            # Should return empty calls without error
+            self.assertIsInstance(result, StreamingParseResult)
+            self.assertEqual(result.calls, [])
+
 
 class TestGlm47MoeDetector(unittest.TestCase):
     def setUp(self):
diff --git a/test/registered/function_call/test_glm47_moe_detector.py b/test/registered/function_call/test_glm47_moe_detector.py
new file mode 100644
index 000000000..a04696406
--- /dev/null
+++ b/test/registered/function_call/test_glm47_moe_detector.py
@@ -0,0 +1,1176 @@
+import json
+import unittest
+
+from sglang.srt.entrypoints.openai.protocol import Function, Tool
+from sglang.srt.function_call.core_types import StreamingParseResult
+from sglang.srt.function_call.glm47_moe_detector import Glm47MoeDetector
+from sglang.test.ci.ci_register import register_cpu_ci
+
+register_cpu_ci(1.0, "default")
+
+
+class TestGlm47MoeDetector(unittest.TestCase):
+    def setUp(self):
+        self.tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="get_weather",
+                    description="Get weather information",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "city": {"type": "string", "description": "City name"},
+                            "date": {"type": "string", "description": "Date"},
+                        },
+                        "required": ["city", "date"],
+                    },
+                ),
+            ),
+        ]
+        self.detector = Glm47MoeDetector()
+
+    # ==================== Basic Parsing Tests (5) ====================
+
+    def test_single_tool_call(self):
+        """
+        Test basic single tool call parsing.
+
+        Scenario: Parse a complete tool call with two string parameters in a single text block.
+        Purpose: Verify the detector can correctly identify and extract function name and parameters
+                from a simple, well-formed tool call.
+        """
+        text = (
+            "<tool_call>get_weather"
+            "<arg_key>city</arg_key><arg_value>Beijing</arg_value>"
+            "<arg_key>date</arg_key><arg_value>2024-06-27</arg_value>"
+            "</tool_call>"
+        )
+        result = self.detector.detect_and_parse(text, self.tools)
+        self.assertEqual(len(result.calls), 1)
+        self.assertEqual(result.calls[0].name, "get_weather")
+        self.assertEqual(
+            result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
+        )
+        self.assertEqual(result.normal_text, "")
+
+    def test_multiple_tool_calls(self):
+        """
+        Test parsing multiple consecutive tool calls.
+
+        Scenario: Parse two complete tool calls back-to-back without any text in between.
+        Purpose: Verify the detector correctly handles multiple tool calls and resets state
+                between calls to avoid parameter leakage or ID conflicts.
+        """
+        text = (
+            "<tool_call>get_weather"
+            "<arg_key>city</arg_key><arg_value>Beijing</arg_value>"
+            "<arg_key>date</arg_key><arg_value>2024-06-27</arg_value>"
+            "</tool_call>"
+            "<tool_call>get_weather"
+            "<arg_key>city</arg_key><arg_value>Shanghai</arg_value>"
+            "<arg_key>date</arg_key><arg_value>2024-06-28</arg_value>"
+            "</tool_call>"
+        )
+        result = self.detector.detect_and_parse(text, self.tools)
+        self.assertEqual(len(result.calls), 2)
+        self.assertEqual(result.calls[0].name, "get_weather")
+        self.assertEqual(
+            result.calls[0].parameters, '{"city": "Beijing", "date": "2024-06-27"}'
+        )
+        self.assertEqual(result.calls[1].name, "get_weather")
+        self.assertEqual(
+            result.calls[1].parameters, '{"city": "Shanghai", "date": "2024-06-28"}'
+        )
+        self.assertEqual(result.normal_text, "")
+
+    def test_no_arg_function_non_streaming(self):
+        """
+        Test no-argument function call without streaming.
+
+        Scenario: Parse a tool call for a function that has no parameters (empty properties).
+        Purpose: Verify the detector generates a single empty object "{}" for no-argument functions
+                and does not duplicate empty parameter objects.
+        """
+        tools_with_no_args = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="list_filenames",
+                    description="List filenames",
+                    parameters={
+                        "type": "object",
+                        "properties": {},
+                    },
+                ),
+            ),
+        ]
+
+        text = "<tool_call>list_filenames</tool_call>"
+        result = self.detector.detect_and_parse(text, tools_with_no_args)
+
+        self.assertEqual(len(result.calls), 1)
+        self.assertEqual(result.calls[0].name, "list_filenames")
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params, {})
+
+    def test_invalid_tool_call(self):
+        """
+        Test handling of invalid tool calls.
+
+        Scenario: Attempt to parse a tool call with a function name that doesn't exist in the tool list.
+        Purpose: Verify the detector gracefully rejects invalid function calls and returns no calls
+                rather than throwing an error or accepting invalid input.
+        """
+        text = "<tool_call>invalid_func<arg_key>city</arg_key><arg_value>Beijing</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, self.tools)
+        self.assertEqual(len(result.calls), 0)
+
+    def test_array_argument_with_escaped_json(self):
+        """
+        Test array arguments containing escaped JSON strings.
+
+        Scenario: Parse tool calls with array parameters containing nested JSON objects with
+                 escaped quotes (both backslash-escaped and raw escaped strings).
+        Purpose: Verify the detector properly handles JSON escaping without double-escaping,
+                preserving special characters like backslashes in paths and newline sequences.
+        """
+        tools_with_array = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="todo_write",
+                    description="Write todos",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "todos": {
+                                "type": "array",
+                                "description": "The updated todo list",
+                            }
+                        },
+                        "required": ["todos"],
+                    },
+                ),
+            ),
+        ]
+
+        def check_params(result):
+            self.assertEqual(1, len(result.calls))
+            self.assertEqual("todo_write", result.calls[0].name)
+            params = json.loads(result.calls[0].parameters)
+            self.assertIsInstance(params["todos"], list)
+            self.assertEqual(4, len(params["todos"]))
+            self.assertEqual("1", params["todos"][0]["id"])
+            self.assertEqual(
+                "Check for hard-coded issues in the backend code",
+                params["todos"][0]["task"],
+            )
+            self.assertEqual("in_progress", params["todos"][0]["status"])
+            self.assertEqual("2", params["todos"][1]["id"])
+            self.assertEqual(
+                "Check for hard-coded issues in the frontend code",
+                params["todos"][1]["task"],
+            )
+            self.assertEqual("pending", params["todos"][1]["status"])
+            self.assertEqual("3", params["todos"][2]["id"])
+            self.assertEqual(
+                "Check for code violating the Single Responsibility Principle",
+                params["todos"][2]["task"],
+            )
+            self.assertEqual("pending", params["todos"][2]["status"])
+            self.assertEqual("4", params["todos"][3]["id"])
+            self.assertEqual(
+                "Generate a rectification proposal report", params["todos"][3]["task"]
+            )
+            self.assertEqual("pending", params["todos"][3]["status"])
+
+        # Test with normal escaped JSON in XML
+        result = self.detector.detect_and_parse(
+            """<tool_call>todo_write<arg_key>todos</arg_key><arg_value>[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]</arg_value>
+</tool_call>""",
+            tools_with_array,
+        )
+        check_params(result)
+
+        # Test with raw string escaped JSON
+        result = self.detector.detect_and_parse(
+            r"""<tool_call>todo_write<arg_key>todos</arg_key><arg_value>[{\"id\": \"1\", \"task\": \"Check for hard-coded issues in the backend code\", \"status\": \"in_progress\"}, {\"id\": \"2\", \"task\": \"Check for hard-coded issues in the frontend code\", \"status\": \"pending\"}, {\"id\": \"3\", \"task\": \"Check for code violating the Single Responsibility Principle\", \"status\": \"pending\"}, {\"id\": \"4\", \"task\": \"Generate a rectification proposal report\", \"status\": \"pending\"}]</arg_value>
+</tool_call>""",
+            tools_with_array,
+        )
+        check_params(result)
+
+        def check_single_todos(tool_result, expected):
+            self.assertEqual(1, len(tool_result.calls))
+            self.assertEqual("todo_write", tool_result.calls[0].name)
+            params = json.loads(tool_result.calls[0].parameters)
+            self.assertIsInstance(params["todos"], list)
+            self.assertEqual(1, len(params["todos"]))
+            self.assertEqual("1", params["todos"][0]["id"])
+            self.assertEqual(expected, params["todos"][0]["task"])
+            self.assertEqual("pending", params["todos"][0]["status"])
+
+        # Test with escaped backslashes (Windows paths)
+        expected_path = r"Check file at C:\Users\test.txt"
+        result = self.detector.detect_and_parse(
+            """<tool_call>todo_write<arg_key>todos</arg_key><arg_value>[{\"id\": \"1\", \"task\": \"Check file at C:\\\\Users\\\\test.txt\", \"status\": \"pending\"}]</arg_value></tool_call>""",
+            tools_with_array,
+        )
+        check_single_todos(result, expected_path)
+
+        # Test with literal backslash-n (not newline)
+        expected_output = r"Print \n to see newline"
+        result = self.detector.detect_and_parse(
+            """<tool_call>todo_write<arg_key>todos</arg_key><arg_value>[{\"id\": \"1\", \"task\": \"Print \\\\n to see newline\",\"status\": \"pending\"}]</arg_value></tool_call>""",
+            tools_with_array,
+        )
+        check_single_todos(result, expected_output)
+
+    # ==================== MTP Core Scenarios (3) ====================
+
+    def test_mtp_func_and_string_split(self):
+        """
+        Test MTP-style function name and string parameter value splitting across chunks.
+
+        Scenario: Simulate Model Token Provider (MTP) behavior where function names and string
+                 parameter values are split mid-word across multiple chunks.
+        Purpose: This is the MOST CRITICAL test - verify the detector correctly reassembles:
+                - Function name split as "create_ta" + "sk"
+                - String values split as "Go to Bei" + "jing" and "San Fran" + "cisco"
+                These splits mimic real MTP output where tokenization breaks words arbitrarily.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="create_task",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "title": {"type": "string"},
+                            "location": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        chunks = [
+            "I'll create a task.",  # normal text before tool call
+            "<tool_call>create_ta",  # function name split mid-word
+            "sk<arg_key>title</arg_key><arg_value>Go to Bei",  # function name completes, param value splits
+            "jing</arg_value>",  # first parameter value completes
+            "<arg_key>location</arg_key><arg_value>San Fran",  # second parameter value splits
+            "cisco</arg_value></tool_call>",  # second parameter and tool call complete
+        ]
+
+        detector = Glm47MoeDetector()
+        all_calls = []
+        all_normal_text = ""
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, tools)
+            all_calls.extend(result.calls)
+            all_normal_text += result.normal_text
+
+        # Verify normal text is preserved
+        self.assertEqual(all_normal_text, "I'll create a task.")
+
+        # Verify function call
+        func_calls = [c for c in all_calls if c.name]
+        self.assertEqual(len(func_calls), 1)
+        self.assertEqual(
+            func_calls[0].name, "create_task"
+        )  # "create_ta" + "sk" reassembled
+
+        # Verify parameter reassembly
+        full_params = "".join([c.parameters for c in all_calls if c.parameters])
+        params = json.loads(full_params)
+        self.assertEqual(
+            params["title"], "Go to Beijing"
+        )  # "Go to Bei" + "jing" reassembled
+        self.assertEqual(
+            params["location"], "San Francisco"
+        )  # "San Fran" + "cisco" reassembled
+
+    def test_mtp_noarg_and_multiple_calls(self):
+        """
+        Test MTP-style no-argument function and multiple tool calls with state reset.
+
+        Scenario: Stream a no-argument function call followed by a regular function call,
+                 simulating MTP's output pattern where function completion triggers state reset.
+        Purpose: Verify:
+                - No-argument functions emit exactly ONE empty object "{}", not duplicates
+                - State properly resets between consecutive tool calls (tool_index increments)
+                - Second tool call doesn't inherit parameters from first call
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="list_files",
+                    parameters={
+                        "type": "object",
+                        "properties": {},
+                    },
+                ),
+            ),
+            Tool(
+                type="function",
+                function=Function(
+                    name="get_weather",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "city": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        chunks = [
+            "<tool_call>list_files</tool_call>",  # no-arg function, complete in one chunk
+            "<tool_call>get_weather<arg_key>city</arg_key><arg_value>Beijing</arg_value></tool_call>",
+        ]
+
+        detector = Glm47MoeDetector()
+        all_calls = []
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, tools)
+            all_calls.extend(result.calls)
+
+        # Verify two distinct tool calls
+        func_calls = [c for c in all_calls if c.name]
+        self.assertEqual(len(func_calls), 2)
+        self.assertEqual(func_calls[0].name, "list_files")
+        self.assertEqual(func_calls[1].name, "get_weather")
+
+        # Verify no duplicate empty objects for no-arg function
+        empty_object_calls = [c for c in all_calls if c.parameters == "{}"]
+        self.assertLessEqual(
+            len(empty_object_calls),
+            1,
+            "No-argument function should emit at most one empty object",
+        )
+
+        # Verify second call has correct parameters
+        weather_params = [
+            c.parameters for c in all_calls if c.parameters and c.parameters != "{}"
+        ]
+        if weather_params:
+            full_params = "".join(weather_params)
+            params = json.loads(full_params)
+            self.assertEqual(params["city"], "Beijing")
+
+    def test_mtp_number_and_complex_json(self):
+        """
+        Test MTP-style number parameters and complex JSON array splitting.
+
+        Scenario: Parse tool calls with number parameters (int and float) and JSON arrays
+                 split across chunks, including splits within JSON structure.
+        Purpose: Verify:
+                - Number types (5.5, 10) are preserved as numbers, not strings
+                - JSON array content split as "description" + ": \"" maintains validity
+                - Nested JSON objects in arrays are correctly reconstructed
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="create_todos",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "priority": {"type": "number"},
+                            "count": {"type": "integer"},
+                            "items": {"type": "array"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        chunks = [
+            "<tool_call>create_todos",
+            "<arg_key>priority</arg_key><arg_value>5.5</arg_value>",  # float number
+            "<arg_key>count</arg_key><arg_value>10</arg_value>",  # integer number
+            '<arg_key>items</arg_key><arg_value>[{"description',  # JSON array splits mid-key
+            '": "Test',  # key completes, value starts
+            'Todo 1"}, {"description": "TestTodo 2"}]</arg_value></tool_call>',
+        ]
+
+        detector = Glm47MoeDetector()
+        all_calls = []
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, tools)
+            all_calls.extend(result.calls)
+
+        # Verify function name
+        func_calls = [c for c in all_calls if c.name]
+        self.assertEqual(len(func_calls), 1)
+        self.assertEqual(func_calls[0].name, "create_todos")
+
+        # Verify parameters - numbers and JSON array
+        full_params = "".join([c.parameters for c in all_calls if c.parameters])
+        params = json.loads(full_params)
+
+        # Number types should be preserved
+        self.assertIsInstance(params["priority"], (int, float))
+        self.assertEqual(params["priority"], 5.5)
+        self.assertIsInstance(params["count"], int)
+        self.assertEqual(params["count"], 10)
+
+        # JSON array should be correctly reconstructed
+        self.assertIsInstance(params["items"], list)
+        self.assertEqual(len(params["items"]), 2)
+        self.assertEqual(params["items"][0]["description"], "TestTodo 1")
+        self.assertEqual(params["items"][1]["description"], "TestTodo 2")
+
+    # ==================== Streaming Basics (3) ====================
+
+    def test_streaming_tool_call(self):
+        """
+        Test basic streaming incremental parsing of a single tool call.
+
+        Scenario: Parse a tool call split across 4 chunks with natural boundaries
+                 (function name, first param, second param, closing tag).
+        Purpose: Verify basic streaming functionality works correctly and accumulates
+                parameters progressively across chunks.
+        """
+        chunks = [
+            "<tool_call>get_weather",
+            "<arg_key>city</arg_key><arg_value>Beijing</arg_value>",
+            "<arg_key>date</arg_key><arg_value>2024-06-27</arg_value>",
+            "</tool_call>",
+        ]
+        tool_calls = []
+        for chunk in chunks:
+            result = self.detector.parse_streaming_increment(chunk, self.tools)
+            for tool_call_chunk in result.calls:
+                if (
+                    hasattr(tool_call_chunk, "tool_index")
+                    and tool_call_chunk.tool_index is not None
+                ):
+                    while len(tool_calls) <= tool_call_chunk.tool_index:
+                        tool_calls.append({"name": "", "parameters": ""})
+                    tc = tool_calls[tool_call_chunk.tool_index]
+                    if tool_call_chunk.name:
+                        tc["name"] = tool_call_chunk.name
+                    if tool_call_chunk.parameters:
+                        tc["parameters"] += tool_call_chunk.parameters
+        self.assertEqual(len(tool_calls), 1)
+        self.assertEqual(tool_calls[0]["name"], "get_weather")
+        self.assertEqual(
+            tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
+        )
+
+    def test_streaming_multiple_tool_calls(self):
+        """
+        Test streaming incremental parsing of multiple consecutive tool calls.
+
+        Scenario: Stream two complete tool calls with the transition "</tool_call><tool_call>"
+                 occurring within a single chunk.
+        Purpose: Verify streaming correctly handles multiple tool calls and properly increments
+                tool_index for each new call.
+        """
+        chunks = [
+            "<tool_call>get_weather",
+            "<arg_key>city</arg_key><arg_value>Beijing</arg_value>",
+            "<arg_key>date</arg_key><arg_value>2024-06-27</arg_value>",
+            "</tool_call><tool_call>get_weather",  # two tool calls transition in same chunk
+            "<arg_key>city</arg_key><arg_value>Shanghai</arg_value>",
+            "<arg_key>date</arg_key><arg_value>2024-06-28</arg_value>",
+            "</tool_call>",
+        ]
+        tool_calls = []
+        for chunk in chunks:
+            result = self.detector.parse_streaming_increment(chunk, self.tools)
+            for tool_call_chunk in result.calls:
+                if (
+                    hasattr(tool_call_chunk, "tool_index")
+                    and tool_call_chunk.tool_index is not None
+                ):
+                    while len(tool_calls) <= tool_call_chunk.tool_index:
+                        tool_calls.append({"name": "", "parameters": ""})
+                    tc = tool_calls[tool_call_chunk.tool_index]
+                    if tool_call_chunk.name:
+                        tc["name"] = tool_call_chunk.name
+                    if tool_call_chunk.parameters:
+                        tc["parameters"] += tool_call_chunk.parameters
+        self.assertEqual(len(tool_calls), 2)
+        self.assertEqual(tool_calls[0]["name"], "get_weather")
+        self.assertEqual(
+            tool_calls[0]["parameters"], '{"city": "Beijing", "date": "2024-06-27"}'
+        )
+        self.assertEqual(tool_calls[1]["name"], "get_weather")
+        self.assertEqual(
+            tool_calls[1]["parameters"], '{"city": "Shanghai", "date": "2024-06-28"}'
+        )
+
+    def test_normal_text_before_tool_call(self):
+        """
+        Test preservation of normal text (including punctuation) before tool calls.
+
+        Scenario: Parse chunks containing normal text with various punctuation marks
+                 (English and Chinese) immediately followed by tool call tags.
+        Purpose: Verify normal text is preserved in result.normal_text and not lost when
+                tool call parsing begins. This consolidates 6 previous Chinese punctuation tests.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="list_dir",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "path": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        test_cases = [
+            ("Sure, let me help.<tool_call>list_dir", "English with period"),
+            ("结构：<tool_call>list_dir", "Chinese colon"),
+            ("问题。<tool_call>list_dir", "Chinese period"),
+            ("Complete!<tool_call>list_dir", "English exclamation"),
+            ("说明；<tool_call>list_dir", "Chinese semicolon"),
+        ]
+
+        for text, description in test_cases:
+            with self.subTest(description=description):
+                detector = Glm47MoeDetector()
+                result = detector.parse_streaming_increment(text, tools)
+
+                before_token = text.split("<tool_call>")[0]
+                self.assertIn(
+                    before_token,
+                    result.normal_text,
+                    f"Should preserve '{before_token}' in '{description}'",
+                )
+
+    # ==================== Boundary Cases (9) ====================
+
+    def test_boundary_empty_param_value(self):
+        """
+        Test handling of empty parameter values.
+
+        Scenario: Parse a tool call where a parameter value is an empty string.
+        Purpose: Verify the detector correctly handles empty strings as valid parameter values
+                and doesn't skip or error on them.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="create_note",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "title": {"type": "string"},
+                            "content": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        text = "<tool_call>create_note<arg_key>title</arg_key><arg_value>Test</arg_value><arg_key>content</arg_key><arg_value></arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+
+        self.assertEqual(len(result.calls), 1)
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["title"], "Test")
+        self.assertEqual(params["content"], "")  # empty string should be preserved
+
+    def test_boundary_param_value_extreme_split(self):
+        """
+        Test extreme parameter value splitting - one character per chunk.
+
+        Scenario: Stream a parameter value where each character arrives in a separate chunk,
+                 representing worst-case MTP tokenization.
+        Purpose: Stress test the buffer reassembly mechanism to ensure it can handle
+                extremely granular chunk boundaries without data loss or corruption.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="search",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        chunks = [
+            "<tool_call>search<arg_key>query</arg_key><arg_value>N",
+            "e",
+            "w ",
+            "Y",
+            "o",
+            "rk</arg_value></tool_call>",
+        ]
+
+        detector = Glm47MoeDetector()
+        all_calls = []
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, tools)
+            all_calls.extend(result.calls)
+
+        full_params = "".join([c.parameters for c in all_calls if c.parameters])
+        params = json.loads(full_params)
+        self.assertEqual(
+            params["query"], "New York"
+        )  # all characters correctly reassembled
+
+    def test_boundary_param_value_with_special_chars(self):
+        """
+        Test parameter values containing special characters and escape sequences.
+
+        Scenario: Parse parameter values with quotes, backslashes, newlines, and other
+                 special characters that require JSON escaping.
+        Purpose: Verify special characters are properly escaped/unescaped and preserved
+                through the parsing pipeline without corruption.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="execute_command",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "command": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # Test with single quotes (no escaping needed)
+        text = "<tool_call>execute_command<arg_key>command</arg_key><arg_value>echo 'Hello World'</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["command"], "echo 'Hello World'")
+
+        # Test with spaces and special chars that don't need escaping
+        text = "<tool_call>execute_command<arg_key>command</arg_key><arg_value>echo Hello & World</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["command"], "echo Hello & World")
+
+    def test_boundary_json_deeply_nested(self):
+        """
+        Test deeply nested JSON structures in parameter values.
+
+        Scenario: Parse a parameter containing a deeply nested JSON object with multiple levels.
+        Purpose: Verify the detector can handle complex nested structures without stack overflow
+                or parsing errors.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="process_data",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "data": {"type": "object"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        nested_json = (
+            '{"level1": {"level2": {"level3": {"level4": {"value": "deep"}}}}}'
+        )
+        text = f"<tool_call>process_data<arg_key>data</arg_key><arg_value>{nested_json}</arg_value></tool_call>"
+
+        result = self.detector.detect_and_parse(text, tools)
+        params = json.loads(result.calls[0].parameters)
+
+        # Navigate through nested structure
+        self.assertEqual(
+            params["data"]["level1"]["level2"]["level3"]["level4"]["value"], "deep"
+        )
+
+    def test_boundary_json_empty_structures(self):
+        """
+        Test empty JSON structures (empty objects and arrays) in parameters.
+
+        Scenario: Parse parameters containing empty objects {} and empty arrays [].
+        Purpose: Verify empty structures are preserved and not confused with no-argument
+                function empty parameter generation.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="create_structure",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "empty_obj": {"type": "object"},
+                            "empty_arr": {"type": "array"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        text = "<tool_call>create_structure<arg_key>empty_obj</arg_key><arg_value>{}</arg_value><arg_key>empty_arr</arg_key><arg_value>[]</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["empty_obj"], {})
+        self.assertEqual(params["empty_arr"], [])
+
+    def test_boundary_multi_tags_one_chunk(self):
+        """
+        Test multiple XML tags appearing in a single chunk.
+
+        Scenario: Parse chunks where multiple complete tags (arg_key, arg_value, etc.)
+                 appear together without any chunk boundaries between them.
+        Purpose: Verify the regex-based tag extraction correctly handles multiple tags
+                in one chunk and processes them in the correct order.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="multi_param",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "a": {"type": "string"},
+                            "b": {"type": "string"},
+                            "c": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # All three parameters in one chunk
+        text = "<tool_call>multi_param<arg_key>a</arg_key><arg_value>1</arg_value><arg_key>b</arg_key><arg_value>2</arg_value><arg_key>c</arg_key><arg_value>3</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["a"], "1")
+        self.assertEqual(params["b"], "2")
+        self.assertEqual(params["c"], "3")
+
+    def test_boundary_normal_text_mixed_with_tool(self):
+        """
+        Test normal text interleaved with tool calls.
+
+        Scenario: Parse text with normal text before and after tool calls.
+        Purpose: Verify normal text segments are correctly separated from tool call parsing
+                and preserved in the normal_text output.
+
+        NOTE: Currently, the detector only captures text BEFORE the first tool call.
+              Text after tool calls is not returned (known limitation).
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="action",
+                    parameters={
+                        "type": "object",
+                        "properties": {},
+                    },
+                ),
+            ),
+        ]
+
+        text = "First I'll do this.<tool_call>action</tool_call>Then I'll do that."
+        result = self.detector.detect_and_parse(text, tools)
+
+        self.assertEqual(len(result.calls), 1)
+        self.assertEqual(result.calls[0].name, "action")
+        # Currently only text before tool call is captured
+        self.assertIn("First I'll do this.", result.normal_text)
+        # TODO: Text after tool call should also be preserved but currently isn't
+        self.assertIn("Then I'll do that.", result.normal_text)
+
+    def test_boundary_number_edge_values(self):
+        """
+        Test edge-case number values (zero, negative, scientific notation).
+
+        Scenario: Parse parameters with various numeric edge cases to ensure proper type handling.
+        Purpose: Verify the detector correctly preserves number types for edge values and doesn't
+                convert them to strings or lose precision.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="calculate",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "zero": {"type": "number"},
+                            "negative": {"type": "number"},
+                            "large": {"type": "number"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        text = "<tool_call>calculate<arg_key>zero</arg_key><arg_value>0</arg_value><arg_key>negative</arg_key><arg_value>-42.5</arg_value><arg_key>large</arg_key><arg_value>1e10</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+
+        params = json.loads(result.calls[0].parameters)
+        self.assertEqual(params["zero"], 0)
+        self.assertEqual(params["negative"], -42.5)
+        self.assertEqual(params["large"], 1e10)
+
+    def test_boundary_type_string_with_numeric_content(self):
+        """
+        Test string parameters that contain numeric-looking content.
+
+        Scenario: Parse string parameters with values like "123" or "45.67" that look like
+                 numbers but should remain strings based on parameter schema.
+        Purpose: Verify type preservation based on schema definition, not content appearance.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="store_data",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "id": {
+                                "type": "string"
+                            },  # string type despite numeric content
+                            "code": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        text = "<tool_call>store_data<arg_key>id</arg_key><arg_value>12345</arg_value><arg_key>code</arg_key><arg_value>67.89</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, tools)
+
+        params = json.loads(result.calls[0].parameters)
+        # Should be strings, not numbers
+        self.assertIsInstance(params["id"], str)
+        self.assertIsInstance(params["code"], str)
+        self.assertEqual(params["id"], "12345")
+        self.assertEqual(params["code"], "67.89")
+
+    # ==================== Error Handling (2) ====================
+
+    def test_error_undefined_tool(self):
+        """
+        Test error handling for undefined tool names.
+
+        Scenario: Attempt to call a function that doesn't exist in the provided tools list.
+        Purpose: Verify the detector gracefully handles undefined tools by returning an empty
+                call list rather than crashing or producing malformed output.
+        """
+        text = "<tool_call>nonexistent_function<arg_key>param</arg_key><arg_value>value</arg_value></tool_call>"
+        result = self.detector.detect_and_parse(text, self.tools)
+
+        # Should not crash, should return empty calls
+        self.assertEqual(len(result.calls), 0)
+
+    def test_error_incomplete_buffer_at_end(self):
+        """
+        Test handling of incomplete tool calls at end of stream.
+
+        Scenario: Streaming ends with an incomplete tool call (e.g., missing closing tag).
+        Purpose: Verify the detector handles incomplete buffers gracefully without throwing
+                exceptions, as streaming may end mid-parse in real scenarios.
+        """
+        chunks = [
+            "<tool_call>get_weather<arg_key>city</arg_key><arg_value>Beijing",
+            # Stream ends here, no closing tags
+        ]
+
+        detector = Glm47MoeDetector()
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, self.tools)
+            # Should not crash
+            self.assertIsInstance(result, StreamingParseResult)
+
+        # Incomplete call should not be in results
+        # (or may be partially present - main thing is no exception)
+
+    # ==================== Streamed Raw Length Bug Tests (3) ====================
+
+    def test_streamed_raw_length_incomplete_xml_tag(self):
+        """
+        Test that _streamed_raw_length is updated even when json_increment is empty.
+
+        Scenario: Stream XML content that is split at an incomplete tag boundary,
+                 causing the state machine to buffer without producing JSON output.
+        Purpose: Verify that _streamed_raw_length is updated regardless of whether
+                json_increment is empty, preventing reprocessing of the same input.
+
+        This tests the bug where:
+        1. raw_increment is extracted from func_args_raw[self._streamed_raw_length:]
+        2. _process_xml_to_json_streaming() returns empty string (buffering state)
+        3. If _streamed_raw_length is NOT updated before the early return,
+           the next call will reprocess the same raw_increment
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="get_weather",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "city": {"type": "string"},
+                            "temperature": {"type": "number"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # Simulate streaming chunks where XML tags are split
+        chunks = [
+            "<tool_call>get_weather",
+            "<arg_key>city</arg_key><arg_value>Bei",  # Split in middle of value
+            "jing</arg_value>",  # Complete the value
+            "<arg_key>temperature</arg_key><arg_value>2",  # Split numeric value
+            "5</arg_value></tool_call>",
+        ]
+
+        detector = Glm47MoeDetector()
+        all_calls = []
+        collected_params = ""
+
+        for i, chunk in enumerate(chunks):
+            result = detector.parse_streaming_increment(chunk, tools)
+            all_calls.extend(result.calls)
+
+            # Collect parameters
+            for call in result.calls:
+                if call.parameters:
+                    collected_params += call.parameters
+
+        # Verify complete parameters were collected without duplication
+        if collected_params:
+            params = json.loads(collected_params)
+            self.assertEqual(params["city"], "Beijing")
+            self.assertEqual(params["temperature"], 25)
+
+            # Critical: Verify no duplicate JSON output due to reprocessing
+            # Count occurrences of "city" key - should appear exactly once
+            city_count = collected_params.count('"city"')
+            self.assertEqual(
+                city_count,
+                1,
+                f"'city' key appears {city_count} times, expected 1. "
+                f"This indicates input reprocessing bug.",
+            )
+
+    def test_streamed_raw_length_tag_split_across_chunks(self):
+        """
+        Test _streamed_raw_length update when tag is split across chunk boundaries.
+
+        Scenario: XML tags themselves are split across chunks (e.g., "<arg_k" + "ey>").
+        Purpose: Verify that even when the state machine is buffering partial tags,
+                _streamed_raw_length is correctly updated to prevent reprocessing.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="search",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "query": {"type": "string"},
+                            "limit": {"type": "integer"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # Split tags in extreme positions
+        chunks = [
+            "<tool_call>search<arg_",  # Split tag name
+            "key>query</arg_key><arg_value>Python progra",  # Complete tag, split value
+            "mming</arg_value><arg_",  # Complete value, split next tag
+            "key>limit</arg_key><arg_value>10</arg_value></tool_call>",
+        ]
+
+        detector = Glm47MoeDetector()
+        all_params = ""
+
+        for chunk in chunks:
+            result = detector.parse_streaming_increment(chunk, tools)
+            for call in result.calls:
+                if call.parameters:
+                    all_params += call.parameters
+
+        # Verify correct reassembly
+        params = json.loads(all_params)
+        self.assertEqual(params["query"], "Python programming")
+        self.assertEqual(params["limit"], 10)
+
+        # Verify no duplication in output
+        query_count = all_params.count('"query"')
+        limit_count = all_params.count('"limit"')
+        self.assertEqual(query_count, 1, "query key duplicated - reprocessing bug")
+        self.assertEqual(limit_count, 1, "limit key duplicated - reprocessing bug")
+
+    def test_streamed_raw_length_buffer_only_partial_tag(self):
+        """
+        Test that _streamed_raw_length updates even when state machine returns empty.
+
+        Scenario: Send increment that is ONLY a partial opening tag that state machine
+                 must buffer completely without producing any JSON output.
+        Purpose: Force json_increment to be empty string to expose the bug where
+                _streamed_raw_length is not updated before early return.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="test_func",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "key1": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # Manually call _process_arguments_streaming to have precise control
+        detector = Glm47MoeDetector()
+        detector.current_tool_id = 0
+        detector.current_tool_name_sent = True
+        detector._reset_streaming_state()
+        detector.streamed_args_for_tool = [""]
+        detector._streamed_raw_length = 0
+
+        # First call: Complete tag that produces JSON output
+        func_args_1 = "<arg_key>key1</arg_key><arg_value>va"
+        result_1 = detector._process_arguments_streaming(
+            "test_func", func_args_1, tools
+        )
+
+        # Should produce JSON output: {"key1": "va (partial)
+        self.assertIsNotNone(result_1)
+        self.assertGreater(len(result_1.parameters), 0)
+        initial_length = detector._streamed_raw_length
+        self.assertEqual(initial_length, len(func_args_1))
+
+        # Second call: Add just partial closing tag - state machine will buffer this
+        # without producing JSON (it's waiting to see if </arg_value> is complete)
+        func_args_2 = func_args_1 + "<"  # Add partial tag
+        result_2 = detector._process_arguments_streaming(
+            "test_func", func_args_2, tools
+        )
+
+        # This is the critical test: if _streamed_raw_length is NOT updated when
+        # json_increment is empty, then detector._streamed_raw_length will still be
+        # at initial_length, and the next call will reprocess the "<" character
+
+        # Check if length was updated (bug test)
+        updated_length = detector._streamed_raw_length
+
+        # BUG: If code has bug, updated_length will equal initial_length
+        # FIXED: If code is correct, updated_length should equal len(func_args_2)
+        self.assertEqual(
+            updated_length,
+            len(func_args_2),
+            "Bug detected: _streamed_raw_length not updated when json_increment is empty. "
+            f"Expected {len(func_args_2)}, got {updated_length}",
+        )
+
+    def test_streamed_raw_length_multiple_empty_returns(self):
+        """
+        Test consecutive chunks that produce empty json_increment.
+
+        Scenario: Multiple consecutive chunks that all result in empty json_increment
+                 as the state machine buffers complex nested structures.
+        Purpose: Verify _streamed_raw_length advances correctly through multiple
+                empty-return cycles without getting stuck or reprocessing.
+        """
+        tools = [
+            Tool(
+                type="function",
+                function=Function(
+                    name="update_settings",
+                    parameters={
+                        "type": "object",
+                        "properties": {
+                            "name": {"type": "string"},
+                            "value": {"type": "string"},
+                        },
+                    },
+                ),
+            ),
+        ]
+
+        # Split XML at positions that may cause state machine buffering
+        chunks = [
+            "<tool_call>update_settings<arg_key>na",  # Split in tag name
+            "me</arg_key><arg_val",  # Complete tag, split next tag
+            "ue>co",  # Complete tag start, split value  # codespell:ignore ue
+            "nf",  # Continue value
+            "ig_v1</arg_value><arg_key>val",  # Complete value, split next key
+            "ue</arg_key><arg_value>ena",  # Complete key name, split value  # codespell:ignore ue
+            "bled</arg_value></tool_call>",  # Complete everything
+        ]
+
+        detector = Glm47MoeDetector()
+        all_params = ""
+
+        for i, chunk in enumerate(chunks):
+            result = detector.parse_streaming_increment(chunk, tools)
+
+            for call in result.calls:
+                if call.parameters:
+                    all_params += call.parameters
+
+        # Verify final output is correct
+        self.assertGreater(len(all_params), 0, "Should have generated some parameters")
+        params = json.loads(all_params)
+        self.assertEqual(params["name"], "config_v1")
+        self.assertEqual(params["value"], "enabled")
+
+        # Verify no duplicate keys due to reprocessing
+        name_count = all_params.count('"name"')
+        value_count = all_params.count('"value"')
+        self.assertEqual(
+            name_count,
+            1,
+            f"'name' appears {name_count} times - indicates reprocessing bug",
+        )
+        self.assertEqual(
+            value_count,
+            1,
+            f"'value' appears {value_count} times - indicates reprocessing bug",
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()