[Bugfix][Frontend][gpt-oss] Recover raw tail when Harmony parser ends non-terminal (#47379)

tests/entrypoints/unit_tests/test_context.py

@@ -74,7 +74,7 @@ class FakeHarmonyParser(HarmonyParser):
         self.reasoning_parser = None
         self.tool_parser = None
         self._chunk_results: list[ChunkResult] = []
-        self._flush_results: list[Segment | None] = []
+        self._flush_results: list[list[Segment]] = []
         self.processed_chunks: list[list[int]] = []
 
     def enqueue_chunk_result(
@@ -89,7 +89,7 @@ class FakeHarmonyParser(HarmonyParser):
             )
         )
 
-    def enqueue_flush_result(self, segment: Segment | None) -> None:
+    def enqueue_flush_result(self, segment: list[Segment]) -> None:
         self._flush_results.append(segment)
 
     def process_chunk(self, token_ids) -> ChunkResult:
@@ -98,10 +98,10 @@ class FakeHarmonyParser(HarmonyParser):
             return self._chunk_results.pop(0)
         return ChunkResult(segments=[], reasoning_token_count=0)
 
-    def flush(self) -> Segment | None:
+    def flush(self) -> list[Segment]:
         if self._flush_results:
             return self._flush_results.pop(0)
-        return None
+        return []
 
 
 def make_harmony_context(
@@ -598,13 +598,21 @@ async def test_streaming_message_synchronization():
         content=[TextContent(text=response_text)],
         recipient=Role.USER,
     )
-    flush_segment = Segment(
-        channel="commentary",
-        recipient=None,
-        delta="",
-        completed_message=message,
-    )
-    parser.enqueue_flush_result(flush_segment)
+    flush_segments = [
+        Segment(
+            channel="final",
+            recipient=None,
+            delta=response_text,
+            completed_message=None,
+        ),
+        Segment(
+            channel="final",
+            recipient=None,
+            delta="",
+            completed_message=message,
+        ),
+    ]
+    parser.enqueue_flush_result(flush_segments)
 
     # Create another output to trigger synchronization via flush()
     context.append_output(
@@ -618,8 +626,9 @@ async def test_streaming_message_synchronization():
     assert context.num_init_messages == 1
     assert context._messages[2].content[0].text == response_text
     assert context.last_append_flush_status is True
-    assert len(context.last_append_segments) == 1
-    assert context.last_append_segments[0].completed_message is message
+    assert len(context.last_append_segments) == 2
+    assert context.last_append_segments[-2].delta == response_text
+    assert context.last_append_segments[-1].completed_message is message
 
 
 def test_turn_metrics_copy_and_reset():

tests/parser/test_harmony.py

@@ -7,7 +7,6 @@ from collections.abc import Sequence
 import pytest
 from openai_harmony import (
     Conversation,
-    HarmonyError,
     Message,
     RenderConversationConfig,
     Role,
@@ -51,6 +50,15 @@ def chat_request():
     )
 
 
+@pytest.fixture
+def malformed_msgs_str() -> list[str]:
+    return [
+        "<|channel|>analysis<|message|>thinking<|end|>",
+        "<|start|>assistant<|channel|>commentary<|message|>thinking<|end|>",
+        '<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>',
+    ]
+
+
 def encode_output(harmony_str: str) -> list[int]:
     return get_encoding().encode(harmony_str, allowed_special="all")
 
@@ -131,13 +139,22 @@ def tool_call_entries(delta_message) -> list[tuple[int, str | None, str | None]]
     ]
 
 
+def assert_parser_is_reset(harmony_parser: HarmonyParser):
+    assert harmony_parser._parser is None
+    assert harmony_parser._num_processed_messages == 0
+    assert harmony_parser._current_message_tokens == []
+
+
 class TestFlush:
     def test_flush(self, harmony_parser):
         harmony_parser.process_chunk(
             encode_output("<|channel|>analysis<|message|>Think")
         )
 
-        flushed = harmony_parser.flush()
+        flushed_segments = harmony_parser.flush()
+        assert flushed_segments is not None
+        assert len(flushed_segments) == 1
+        flushed = flushed_segments[0]
 
         assert flushed is not None
         assert flushed.channel == "analysis"
@@ -145,15 +162,27 @@ class TestFlush:
         assert flushed.delta == ""
         assert flushed.completed_message is not None
         assert get_text(flushed.completed_message) == "Think"
-        assert harmony_parser._parser is None
+        assert_parser_is_reset(harmony_parser)
 
-    def test_flush_raises_and_resets_on_non_terminal_eos(self, harmony_parser):
-        harmony_parser.process_chunk(encode_output("<|channel|>analysis"))
+    def test_flush_recovers_invalid_output(self, harmony_parser, malformed_msgs_str):
+        for msg_str in malformed_msgs_str[:-1]:
+            chunk = harmony_parser.process_chunk(encode_output(msg_str))
+            assert "".join(segment.delta for segment in chunk.segments) == "thinking"
 
-        with pytest.raises(HarmonyError):
-            harmony_parser.flush()
+        last_msg_str = malformed_msgs_str[-1]
+        harmony_parser.process_chunk(encode_output(last_msg_str))
+        flushed_segments = harmony_parser.flush()
+        assert len(flushed_segments) == 2
+        delta_segment = flushed_segments[0]
+        message_segment = flushed_segments[1]
 
-        assert harmony_parser._parser is None
+        assert delta_segment.channel == "final"
+        assert delta_segment.recipient is None
+        assert delta_segment.delta == last_msg_str
+        assert message_segment.channel == "final"
+        assert message_segment.recipient is None
+        assert get_text(message_segment.completed_message) == last_msg_str
+        assert_parser_is_reset(harmony_parser)
 
 
 class TestParse:
@@ -364,7 +393,7 @@ class TestParse:
         assert reasoning is None
         assert content == "I'm in the middle of answering"
         assert tool_calls is None
-        assert harmony_parser._parser is None
+        assert_parser_is_reset(harmony_parser)
 
     def test_interrupted_reasoning_first_message(self, harmony_parser, chat_request):
         reasoning, content, tool_calls = harmony_parser.parse(
@@ -378,7 +407,7 @@ class TestParse:
         assert reasoning == "I'm in the middle of thinking"
         assert content is None
         assert tool_calls is None
-        assert harmony_parser._parser is None
+        assert_parser_is_reset(harmony_parser)
 
     def test_truncated_output(self, harmony_parser, chat_request):
         reasoning, content, tool_calls = harmony_parser.parse(
@@ -394,24 +423,23 @@ class TestParse:
         assert reasoning == "I'm thinking."
         assert content == "I'm in the middle of answering"
         assert tool_calls is None
-        assert harmony_parser._parser is None
+        assert_parser_is_reset(harmony_parser)
 
-    def test_malformed_final_recovers_raw_content(self, harmony_parser, chat_request):
-        raw_output = (
-            "<|channel|>analysis<|message|>thinking<|end|>"
-            '<|start|>assistant<|channel|>final {"answer": "hi"}<|return|>'
-        )
+    def test_malformed_msgs_recovers_raw_content(
+        self, harmony_parser, chat_request, malformed_msgs_str
+    ):
+        combined_output = "".join(malformed_msgs_str)
 
         reasoning, content, tool_calls = harmony_parser.parse(
-            raw_output,
+            "",
             chat_request,
-            model_output_token_ids=encode_output(raw_output),
+            model_output_token_ids=encode_output(combined_output),
         )
 
-        assert content == raw_output
-        assert reasoning is None
+        assert reasoning == "thinking"
+        assert content == "thinking\n" + malformed_msgs_str[-1]
         assert tool_calls is None
-        assert harmony_parser._parser is None
+        assert_parser_is_reset(harmony_parser)
 
     @pytest.mark.parametrize(
         ("harmony_str", "expected_content"),
@@ -489,7 +517,7 @@ class TestParseDelta:
         assert second_delta is not None
         assert second_delta.content == "Answer"
         assert second_delta.reasoning is None
-        assert parser._parser is None
+        assert_parser_is_reset(parser)
 
     def test_multi_token(self, gpt_oss_tokenizer, chat_request):
         parser = HarmonyParser(gpt_oss_tokenizer)
@@ -506,25 +534,33 @@ class TestParseDelta:
         assert delta.reasoning is None
         assert not delta.tool_calls
 
-    def test_malformed_final_recovers_raw_content(
-        self, gpt_oss_tokenizer, chat_request
+    def test_malformed_msgs_recovers_raw_content(
+        self, gpt_oss_tokenizer, chat_request, malformed_msgs_str
     ):
         parser = HarmonyParser(gpt_oss_tokenizer)
 
-        delta = parser.parse_delta(
-            delta_text='final {"answer": "hi"}',
-            delta_token_ids=encode_output(
-                '<|channel|>final {"answer": "hi"}<|return|>'
-            ),
+        for msg_str in malformed_msgs_str[:-1]:
+            delta = parser.parse_delta(
+                delta_text="",
+                delta_token_ids=encode_output(msg_str),
+                request=chat_request,
+                finished=False,
+            )
+            assert delta.reasoning or delta.content == "thinking"
+            assert not delta.tool_calls
+
+        last_delta = parser.parse_delta(
+            delta_text="",
+            delta_token_ids=encode_output(malformed_msgs_str[-1]),
             request=chat_request,
             finished=True,
         )
 
-        assert delta is not None
-        assert delta.content == 'final {"answer": "hi"}'
-        assert delta.reasoning is None
-        assert not delta.tool_calls
-        assert parser._parser is None
+        assert last_delta is not None
+        assert last_delta.content == malformed_msgs_str[-1]
+        assert last_delta.reasoning is None
+        assert not last_delta.tool_calls
+        assert_parser_is_reset(parser)
 
     @pytest.mark.parametrize("tool_channel", ["commentary", "analysis"])
     def test_tool_call_split_across_deltas(

vllm/entrypoints/openai/responses/context.py

@@ -18,7 +18,7 @@ from openai.types.responses.response_output_item import McpCall
 from openai.types.responses.response_output_message import ResponseOutputMessage
 from openai.types.responses.response_output_text import ResponseOutputText
 from openai.types.responses.tool import Mcp
-from openai_harmony import Author, HarmonyError, Message, Role, TextContent
+from openai_harmony import Author, Message, Role, TextContent
 
 from vllm import envs
 from vllm.entrypoints.chat_utils import (
@@ -616,7 +616,7 @@ class HarmonyContext(ConversationContext):
         self.num_tool_output_tokens = 0
 
         self.last_append_segments: list[Segment] = []
-        self.last_append_flush_status: bool | HarmonyError = False
+        self.last_append_flush_status: bool = False
 
         # Turn tracking - replaces multiple individual tracking variables
         self.current_turn_metrics = TurnMetrics()
@@ -643,10 +643,10 @@ class HarmonyContext(ConversationContext):
 
         if output.finished:
             self.finish_reason = output.outputs[0].finish_reason
-            flushed = self.response_parser.flush()
-            if flushed is not None:
-                segments.append(flushed)
-            self.last_append_flush_status = flushed is not None
+            flushed_segments = self.response_parser.flush()
+            if flushed_segments:
+                segments.extend(flushed_segments)
+            self.last_append_flush_status = len(flushed_segments) > 0
             self.all_turn_metrics.append(self.current_turn_metrics.copy())
             self.current_turn_metrics.reset()
 

vllm/parser/harmony.py

@@ -9,7 +9,7 @@ from dataclasses import dataclass
 from enum import Enum, auto
 from typing import TYPE_CHECKING, NamedTuple
 
-from openai_harmony import HarmonyError
+from openai_harmony import HarmonyError, Message, Role
 
 from vllm.entrypoints.chat_utils import make_tool_call_id
 from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionRequest
@@ -91,6 +91,9 @@ class HarmonyParser(DelegatingParser):
         self._next_tool_call_index = 0
         self._num_processed_messages = 0
 
+        # For error recovery
+        self._current_message_tokens: list[int] = []
+
     @property
     def _harmony_parser(self) -> StreamableParser:
         """Lazily initializes the Harmony parser."""
@@ -107,32 +110,48 @@ class HarmonyParser(DelegatingParser):
         self._num_processed_messages += 1
         return msg
 
-    def flush(self) -> Segment | None:
+    def flush(self) -> list[Segment]:
+        segments: list[Segment] = []
         try:
             self._harmony_parser.process_eos()
             msg = self._poll_completed_message()
         except HarmonyError:
             logger.warning(
                 "Harmony parser ended in a non-terminal state; returning the "
-                "raw unparsed output. This usually indicates a malformed "
-                "assistant turn, e.g. a 'final' channel missing the "
-                "<|message|> delimiter."
+                "recovered raw output."
             )
-            raise
-        finally:
-            # Reset to the initial assistant-parser state for the next turn.
-            self._parser = None
-            self._num_processed_messages = 0
+
+            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()
 
         if msg is None:
-            return None
+            return segments
 
-        return Segment(
-            channel=msg.channel,
-            recipient=msg.recipient,
-            delta="",
-            completed_message=msg,
+        segments.append(
+            Segment(
+                channel=msg.channel,
+                recipient=msg.recipient,
+                delta="",
+                completed_message=msg,
+            )
         )
+        return segments
 
     def parse(
         self,
@@ -147,12 +166,9 @@ class HarmonyParser(DelegatingParser):
         Callers must decide whether to surface them.
         """
         result = self.process_chunk(model_output_token_ids)
-        try:
-            flushed_segment = self.flush()
-        except HarmonyError:
-            return None, model_output, None
-        if flushed_segment is not None:
-            result.segments.append(flushed_segment)
+        flushed_segments = self.flush()
+        if flushed_segments:
+            result.segments.extend(flushed_segments)
 
         reasoning_parts: list[str] = []
         content_parts: list[str] = []
@@ -209,13 +225,9 @@ class HarmonyParser(DelegatingParser):
         )
         result = self.process_chunk(delta_token_ids)
         if finished:
-            try:
-                flushed_segment = self.flush()
-            except HarmonyError:
-                self._next_tool_call_index = 0
-                return DeltaMessage(content=delta_text)
-            if flushed_segment is not None:
-                result.segments.append(flushed_segment)
+            flushed_segments = self.flush()
+            if flushed_segments:
+                result.segments.extend(flushed_segments)
         combined_content = ""
         combined_reasoning = ""
         tool_messages: list[DeltaToolCall] = []
@@ -298,6 +310,11 @@ class HarmonyParser(DelegatingParser):
             delta = self._harmony_parser.last_content_delta or ""
             completed_message = self._poll_completed_message()
 
+            if completed_message is not None:
+                self._current_message_tokens.clear()
+            else:
+                self._current_message_tokens.append(token_id)
+
             if channel == "analysis" or (
                 channel == "commentary" and recipient is not None
             ):