diff --git a/tests/parser/engine/test_parser_engine.py b/tests/parser/engine/test_parser_engine.py
index 36258668215b..c1b1172a99ce 100644
--- a/tests/parser/engine/test_parser_engine.py
+++ b/tests/parser/engine/test_parser_engine.py
@@ -25,6 +25,7 @@
     FunctionDefinition,
 )
 from vllm.parser.abstract_parser import DelegatingParser
+from vllm.parser.qwen3 import _qwen3_arg_converter
 from vllm.parser.engine.adapters import make_adapters
 from vllm.parser.engine.events import EventType, SemanticEvent
 from vllm.parser.engine.parser_engine import ParserEngine
@@ -1412,6 +1413,14 @@ def test_safe_arg_prefix(self, json_str, expected):
 # ── Coercion instability regression tests ────────────────────────
 
 
+def _json_converter(raw_args: str, partial: bool) -> str:
+    """Identity converter: round-trips JSON, returns raw string when partial."""
+    try:
+        return json.dumps(json.loads(raw_args), ensure_ascii=False)
+    except (json.JSONDecodeError, ValueError):
+        return raw_args if partial else ""
+
+
 def _growing_kv_converter(raw_args: str, partial: bool) -> str:
     """Converter that produces growing bare values (no delimiter)."""
     params: dict[str, str] = {}
@@ -1495,6 +1504,76 @@ def test_int_partial_value_flip_is_safe(self):
         assert isinstance(parsed["val"], str)
         assert parsed["extra"] == "ok"
 
+    def test_integer_middle_field_not_truncated_by_coercion(self):
+        """Integer middle field must not cause truncation when coerced at flush.
+
+        Three chunks are required: the first triggers tool-name emission
+        (consuming the chunk without calling _compute_arg_delta), the second
+        leaves the integer value complete in streamed_json as a quoted string
+        ("42") while the JSON is still open, and the third closes the JSON.
+        Without the fix, flush-time coercion changes "42"→42, breaking the
+        startswith invariant and silently dropping the rest of the arguments.
+        """
+        tool = _make_tool(
+            "f", {"n": {"type": "integer"}, "s": {"type": "string"}}
+        )
+        engine = _make_engine(_converter_config(_json_converter), tools=[tool])
+        parsed = _run_streaming_tool(
+            engine, "f", ["{", '"n": "42", "s": "h', 'i"}']
+        )
+        assert parsed == {"n": 42, "s": "hi"}
+        assert isinstance(parsed["n"], int)
+
+    def test_boolean_middle_field_not_truncated_by_coercion(self):
+        """Boolean middle field must not cause truncation when coerced at flush."""
+        tool = _make_tool(
+            "f", {"flag": {"type": "boolean"}, "msg": {"type": "string"}}
+        )
+        engine = _make_engine(_converter_config(_json_converter), tools=[tool])
+        parsed = _run_streaming_tool(
+            engine, "f", ["{", '"flag": "true", "msg": "h', 'i"}']
+        )
+        assert parsed == {"flag": True, "msg": "hi"}
+        assert isinstance(parsed["flag"], bool)
+
+    def test_null_middle_field_not_truncated_by_coercion(self):
+        """Null-typed middle field must not cause truncation when coerced at flush."""
+        tool = _make_tool(
+            "f", {"v": {"type": "null"}, "tag": {"type": "string"}}
+        )
+        engine = _make_engine(_converter_config(_json_converter), tools=[tool])
+        parsed = _run_streaming_tool(
+            engine, "f", ["{", '"v": "null", "tag": "h', 'i"}']
+        )
+        assert parsed == {"v": None, "tag": "hi"}
+        assert parsed["v"] is None
+
+    def test_qwen3_partial_xml_close_tag_not_included_in_value(self):
+        """Qwen3 partial XML close tag must not leak into streamed string value.
+
+        When an XML close tag (</parameter>) is split across chunks, the
+        partial tag text must not appear in the streamed string value for
+        the preceding key.
+        """
+        tool = _make_tool(
+            "f", {"n": {"type": "integer"}, "s": {"type": "string"}}
+        )
+        engine = _make_engine(
+            _converter_config(_qwen3_arg_converter), tools=[tool]
+        )
+        parsed = _run_streaming_tool(
+            engine,
+            "f",
+            [
+                "<parameter=",
+                "n>42</parameter><parameter=s>ok</par",
+                "ameter>",
+            ],
+        )
+        assert parsed == {"n": 42, "s": "ok"}
+        assert isinstance(parsed["n"], int)
+        assert parsed["s"] == "ok"
+
 
 _DROP_VOCAB: dict[str, int] = {
     **_VOCAB,
diff --git a/vllm/parser/engine/parser_engine.py b/vllm/parser/engine/parser_engine.py
index fc2d653779ed..8fbe28a0a8f1 100644
--- a/vllm/parser/engine/parser_engine.py
+++ b/vllm/parser/engine/parser_engine.py
@@ -281,13 +281,16 @@ def _coerce_dict(args: dict, properties: dict) -> tuple[dict, bool]:
     def _safe_arg_prefix(json_str: str, string_keys: set[str] | None = None) -> str:
         """Return the prefix of *json_str* up to the last top-level value.
 
-        Middle values (followed by a comma) are stable across streaming
-        ticks and included.  The trailing value is excluded for non-string
-        values because type coercion may change its serialised form between
-        ticks, which would violate the ``startswith(prev)`` prefix invariant.
-        String values for keys in ``string_keys`` are prefix-stable, so stream
-        their unterminated content instead of buffering long arguments until
-        the closing tag arrives.
+        The trailing value is excluded for non-string keys because type
+        coercion may change its serialised form, which would violate the
+        ``startswith(prev)`` prefix invariant.  String values for keys in
+        ``string_keys`` are prefix-stable, so their unterminated content is
+        streamed incrementally.
+
+        Middle (comma-terminated) values for non-string keys are also excluded
+        for the same coercion reason: the prefix is clipped at the start of the
+        first non-string-typed value so that ``_fix_arg_types`` at flush time
+        never rewrites anything already sent to the client.
         """
         last_colon = -1
         last_key: str | None = None
@@ -296,6 +299,9 @@ def _safe_arg_prefix(json_str: str, string_keys: set[str] | None = None) -> str:
         escape = False
         string_start = -1
         depth = 0
+        # Index of the first non-string-typed value start; -1 if none seen yet.
+        non_string_clip: int = -1
+
         for i, c in enumerate(json_str):
             if escape:
                 escape = False
@@ -319,8 +325,24 @@ def _safe_arg_prefix(json_str: str, string_keys: set[str] | None = None) -> str:
                 last_colon = i
                 last_key = pending_key
                 pending_key = None
+                if string_keys is not None and last_key not in string_keys:
+                    # Record where this key's value starts; clip here so the
+                    # value is never partially sent before flush-time coercion.
+                    val_start = i + 1
+                    while val_start < len(json_str) and json_str[val_start] in (
+                        " ", "\t", "\n", "\r"
+                    ):
+                        val_start += 1
+                    if non_string_clip < 0:
+                        non_string_clip = val_start
+
         if last_colon < 0:
             return ""
+
+        # If a non-string-typed key's value has started, clip before it.
+        if non_string_clip >= 0:
+            return json_str[:non_string_clip]
+
         end = last_colon + 1
         while end < len(json_str) and json_str[end] in (" ", "\t", "\n", "\r"):
             end += 1