[Feature] Tolerate malformed Harmony streams.

tests/parser/test_harmony.py

@@ -491,6 +491,73 @@ def test_commentary_with_recipient_excluded(self, harmony_parser, chat_request):
             ("get_weather", json.dumps({"location": "SF"}))
         ]
 
+    @pytest.mark.parametrize(
+        ("harmony_str", "expected_reasoning", "expected_content", "warning_substr"),
+        [
+            (
+                (
+                    "<|channel|>analysis"
+                    "<|message|>Reasoning here.<|end|>"
+                    # Below 'assistant' appears 2 times => stray token.
+                    "<|start|>assistantassistant<|channel|>final"
+                    "<|message|>Final answer.<|end|>"
+                ),
+                "Reasoning here.",
+                None,
+                "Unknown role: assistantassistant",
+            ),
+            (
+                (
+                    "<|channel|>analysis"
+                    "<|message|>Reasoning here.<|end|>"
+                    # Below channel name ('final') is skipped.
+                    "<|start|>assistant<|channel|>"
+                    "<|message|>Final answer.<|end|>"
+                ),
+                "Reasoning here.",
+                None,
+                "channel marker present but no channel value found in header",
+            ),
+            (
+                (
+                    "<|channel|>analysis"
+                    "<|message|>Reasoning here.<|end|>"
+                    "<|start|>assistant<|channel|>final"
+                    "<|message|>Final answer.<|end|>"
+                    # Below 'assistant' appears 2 times => stray token.
+                    "<|start|>assistantassistant<|channel|>final"
+                    "<|message|>Ignored answer.<|end|>"
+                ),
+                "Reasoning here.",
+                "Final answer.",
+                "Unknown role: assistantassistant",
+            ),
+        ],
+    )
+    def test_malformed_stream_returns_partial_parse(
+        self,
+        harmony_parser,
+        chat_request,
+        caplog,
+        harmony_str,
+        expected_reasoning,
+        expected_content,
+        warning_substr,
+    ):
+        """Malformed Harmony output should warn, stop, and keep prior parse."""
+        with caplog.at_level("WARNING"):
+            reasoning, content, tool_calls = harmony_parser.parse(
+                "",
+                chat_request,
+                model_output_token_ids=encode_output(harmony_str),
+            )
+
+        assert reasoning == expected_reasoning
+        assert content == expected_content
+        assert tool_calls is None
+        assert warning_substr in caplog.text
+        assert harmony_parser._parser is None
+
 
 class TestParseDelta:
     def test_basic(self, gpt_oss_tokenizer, chat_request):
@@ -534,6 +601,40 @@ def test_multi_token(self, gpt_oss_tokenizer, chat_request):
         assert delta.reasoning is None
         assert not delta.tool_calls
 
+    def test_malformed_stream_stops_future_chunks(
+        self, gpt_oss_tokenizer, chat_request, caplog
+    ):
+        """Contents should be parsed until a malformed header is encountered."""
+        parser = HarmonyParser(gpt_oss_tokenizer)
+
+        with caplog.at_level("WARNING"):
+            first_delta = parser.parse_delta(
+                delta_text="",
+                delta_token_ids=encode_output(
+                    "<|channel|>analysis"
+                    "<|message|>Thinking.<|end|>"
+                    "<|start|>assistantassistant<|channel|>final"
+                ),
+                request=chat_request,
+                finished=False,
+            )
+            second_delta = parser.parse_delta(
+                delta_text="",
+                delta_token_ids=encode_output(
+                    "<|message|>Final answer.<|end|>"
+                ),
+                request=chat_request,
+                finished=True,
+            )
+
+        assert first_delta is not None
+        assert first_delta.reasoning == "Thinking."
+        assert first_delta.content is None
+        assert not first_delta.tool_calls
+        assert second_delta is None
+        assert "Unknown role: assistantassistant" in caplog.text
+        assert parser._parser is None
+
     def test_malformed_msgs_recovers_raw_content(
         self, gpt_oss_tokenizer, chat_request, malformed_msgs_str
     ):
@@ -857,3 +958,32 @@ def test_multi_boundary(self, harmony_parser):
             ("analysis", "One"),
             ("final", "Two"),
         ]
+
+    def test_malformed_stream_keeps_completed_message_channels(
+        self, harmony_parser, caplog
+    ):
+        """Malformed stream should keep completed messages and log a warning."""
+        with caplog.at_level("WARNING"):
+            result = harmony_parser.process_chunk(
+                encode_output(
+                    "<|channel|>analysis<|message|>Reasoning here.<|end|>"
+                    "<|start|>assistant<|channel|>final"
+                    "<|message|>Final answer.<|end|>"
+                    "<|start|>assistantassistant<|channel|>final"
+                    "<|message|>Ignored answer.<|end|>"
+                )
+            )
+
+        boundary_segments = [
+            segment
+            for segment in result.segments
+            if segment.completed_message is not None
+        ]
+        assert [
+            (segment.completed_message.channel, get_text(segment.completed_message))
+            for segment in boundary_segments
+        ] == [
+            ("analysis", "Reasoning here."),
+            ("final", "Final answer."),
+        ]
+        assert "Unknown role: assistantassistant" in caplog.text

vllm/parser/harmony.py

@@ -90,6 +90,7 @@ def __init__(self, tokenizer, tools=None, *args, **kwargs):
         self._parser: StreamableParser | None = None
         self._next_tool_call_index = 0
         self._num_processed_messages = 0
+        self._encountered_parse_error = False
 
         # For error recovery
         self._current_message_tokens: list[int] = []
@@ -110,8 +111,42 @@ def _poll_completed_message(self) -> Message | None:
         self._num_processed_messages += 1
         return msg
 
+    def _reset_parser_state(self) -> None:
+        # Reset to the initial assistant-parser state for the next turn.
+        self._parser = None
+        self._num_processed_messages = 0
+        self._encountered_parse_error = False
+        self._current_message_tokens.clear()
+
+    def _recover_raw_current_message(self) -> list[Segment]:
+        if not self._current_message_tokens:
+            return []
+
+        final_channel = "final"
+        text = self.model_tokenizer.decode(self._current_message_tokens)
+        msg = Message.from_role_and_content(Role.ASSISTANT, text).with_channel(
+            final_channel
+        )
+        return [
+            Segment(
+                channel=final_channel,
+                recipient=None,
+                delta=text,
+                completed_message=None,
+            ),
+            Segment(
+                channel=msg.channel,
+                recipient=msg.recipient,
+                delta="",
+                completed_message=msg,
+            ),
+        ]
+
     def flush(self) -> list[Segment]:
-        segments: list[Segment] = []
+        if self._encountered_parse_error:
+            self._reset_parser_state()
+            return []
+
         try:
             self._harmony_parser.process_eos()
             msg = self._poll_completed_message()
@@ -120,38 +155,23 @@ def flush(self) -> list[Segment]:
                 "Harmony parser ended in a non-terminal state; returning the "
                 "recovered raw output."
             )
+            segments = self._recover_raw_current_message()
+            self._reset_parser_state()
+            return segments
 
-            final_channel = "final"
-            text = self.model_tokenizer.decode(self._current_message_tokens)
-            segments.append(
-                Segment(
-                    channel=final_channel,
-                    recipient=None,
-                    delta=text,
-                    completed_message=None,
-                )
-            )
-            msg = Message.from_role_and_content(Role.ASSISTANT, text).with_channel(
-                final_channel
-            )
-
-        # Reset to the initial assistant-parser state for the next turn.
-        self._parser = None
-        self._num_processed_messages = 0
-        self._current_message_tokens.clear()
+        self._reset_parser_state()
 
         if msg is None:
-            return segments
+            return []
 
-        segments.append(
+        return [
             Segment(
                 channel=msg.channel,
                 recipient=msg.recipient,
                 delta="",
                 completed_message=msg,
             )
-        )
-        return segments
+        ]
 
     def parse(
         self,
@@ -296,13 +316,22 @@ def parse_delta(
         return delta_message
 
     def process_chunk(self, token_ids: Sequence[int]) -> ChunkResult:
-        if not token_ids:
+        if not token_ids or self._encountered_parse_error:
             return ChunkResult(segments=[], reasoning_token_count=0)
 
         segments: list[Segment] = []
         reasoning_token_count = 0
         for token_id in token_ids:
-            self._harmony_parser.process(token_id)
+            try:
+                self._harmony_parser.process(token_id)
+            except HarmonyError as err:
+                logger.warning(
+                    "Harmony parser error at token ID %d, returning partial parse: %r",
+                    token_id,
+                    err,
+                )
+                self._encountered_parse_error = True
+                break
             channel = self._harmony_parser.current_channel
             recipient = self._normalize_recipient(
                 self._harmony_parser.current_recipient