From 4662e9d256c2bc9f63bb2bd14ea5502799b856ed Mon Sep 17 00:00:00 2001 From: ankrovv Date: Tue, 30 Jun 2026 21:27:47 -0700 Subject: [PATCH 1/3] fix(gpt-oss): recover Harmony output after missing message delimiter Inject the missing Harmony message delimiter only for visible final/commentary channel headers, preserving well-formed and tool-call streams. Add parser and streaming regression coverage. Assisted-by: OpenAI Codex --- .../openai/parser/test_harmony_utils.py | 138 ++++++++++++++++++ .../openai/parser/harmony_utils.py | 104 ++++++++++++- 2 files changed, 241 insertions(+), 1 deletion(-) diff --git a/tests/entrypoints/openai/parser/test_harmony_utils.py b/tests/entrypoints/openai/parser/test_harmony_utils.py index e08fc472f09d..ff6cb0bce530 100644 --- a/tests/entrypoints/openai/parser/test_harmony_utils.py +++ b/tests/entrypoints/openai/parser/test_harmony_utils.py @@ -13,11 +13,13 @@ create_tool_definition, extract_function_from_recipient, get_encoding, + get_streamable_parser_for_assistant, get_system_message, has_custom_tools, is_function_recipient, parse_chat_input_to_harmony_message, parse_chat_output, + parse_output_into_messages, ) from vllm.entrypoints.openai.responses.harmony import ( response_input_to_harmony, @@ -1199,3 +1201,139 @@ def test_reasoning_with_empty_content_returns_none(self): msg = response_input_to_harmony(item, prev_responses=[]) assert msg is None + + +def _tok(text: str) -> list[int]: + return get_encoding().encode(text, allowed_special="all") + + +# Harmony control tokens used to assemble test streams. +_CHANNEL = _tok("<|channel|>")[0] +_MESSAGE = _tok("<|message|>")[0] +_END = _tok("<|end|>")[0] +_START = _tok("<|start|>")[0] +_RETURN = _tok("<|return|>")[0] + + +def _analysis(text: str) -> list[int]: + return [_CHANNEL, *_tok("analysis"), _MESSAGE, *_tok(text), _END] + + +def _assistant_header() -> list[int]: + return [_START, *_tok("assistant")] + + +def _streamed_final(tokens: list[int]) -> str | None: + """Mirror the streaming chat path: feed tokens incrementally and accumulate + the visible content deltas from the parser.""" + parser = get_streamable_parser_for_assistant() + out = "" + for token in tokens: + parser.process(token) + if parser.current_channel in ("final", "commentary") and ( + not parser.current_recipient + ): + out += parser.last_content_delta or "" + return out or None + + +class TestRepairUnterminatedVisibleChannel: + """A final/commentary header emitted without the ``<|message|>`` delimiter is + silently dropped by the base parser. get_streamable_parser_for_assistant + returns a parser that repairs it, so every path (non-streaming chat, the + openai tool parser, streaming chat, and the Responses API -- all of which + build their parser through that helper) recovers the content.""" + + def test_malformed_final_recovers_on_all_paths(self): + # ``<|channel|>final {"answer": "hi"}<|return|>`` with no ``<|message|>``. + tokens = [ + *_analysis("thinking"), + *_assistant_header(), + _CHANNEL, + *_tok("final"), + *_tok(' {"answer": "hi"}'), + _RETURN, + ] + # Non-streaming chat (parse_chat_output). + reasoning, content, _ = parse_chat_output(tokens) + assert content is not None and "answer" in content + assert reasoning == "thinking" + + # Shared parser (also what the openai tool parser reads). + finals = [ + m + for m in parse_output_into_messages(tokens).messages + if m.channel == "final" + ] + assert finals and "answer" in finals[0].content[0].text + + # Streaming chat. + streamed = _streamed_final(tokens) + assert streamed is not None and "answer" in streamed + + def test_malformed_commentary_preamble_recovers_content(self): + tokens = [ + *_analysis("thinking"), + *_assistant_header(), + _CHANNEL, + *_tok("commentary"), + *_tok(" Here is a summary."), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content is not None and "summary" in content + assert _streamed_final(tokens) is not None + + def test_no_space_after_channel_recovers_correctly(self): + # ``final{...}`` (no separator) -- the token-level boundary must still be + # the channel name, not ``final{``. + tokens = [ + *_analysis("x"), + *_assistant_header(), + _CHANNEL, + *_tok("final"), + *_tok('{"answer": "hi"}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content == '{"answer": "hi"}' + + def test_well_formed_final_is_noop(self): + tokens = [ + *_analysis("thinking"), + *_assistant_header(), + _CHANNEL, + *_tok("final"), + _MESSAGE, + *_tok('{"answer": "hi"}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content == '{"answer": "hi"}' + assert _streamed_final(tokens) == '{"answer": "hi"}' + + def test_analysis_only_has_no_visible_content(self): + tokens = [_CHANNEL, *_tok("analysis"), _MESSAGE, *_tok("reasoning"), _RETURN] + _, content, _ = parse_chat_output(tokens) + assert content is None + + def test_empty_final_stays_empty(self): + # First message: the parser is pre-seeded with the assistant role, so no + # leading ``<|start|>`` is needed. + tokens = [_CHANNEL, *_tok("final"), _MESSAGE, _RETURN] + _, content, _ = parse_chat_output(tokens) + assert content is None + + def test_tool_call_commentary_is_not_treated_as_content(self): + # ``commentary`` with a recipient is a tool call, not visible content; + # the repair must not fire. + tokens = [ + _CHANNEL, + *_tok("commentary"), + *_tok(" to=functions.get_weather"), + _MESSAGE, + *_tok('{"city": "Paris"}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content is None diff --git a/vllm/entrypoints/openai/parser/harmony_utils.py b/vllm/entrypoints/openai/parser/harmony_utils.py index dc62f3c8b749..d4f111fa7227 100644 --- a/vllm/entrypoints/openai/parser/harmony_utils.py +++ b/vllm/entrypoints/openai/parser/harmony_utils.py @@ -462,8 +462,110 @@ def render_for_completion(messages: list[Message]) -> list[int]: return token_ids +class _HarmonyControlTokens: + """Cached Harmony control-token ids used to repair a missing delimiter. + + Resolved lazily from the encoding so there are no magic token numbers. + """ + + def __init__(self) -> None: + enc = get_encoding() + self._enc = enc + + def tid(text: str) -> int: + return enc.encode(text, allowed_special="all")[0] + + self.channel = tid("<|channel|>") + self.message = tid("<|message|>") + self.constrain = tid("<|constrain|>") + self.recipient = tid(" to") # start of a ` to=` tool target + # Tokens that terminate or restart a header; content never starts with one. + self.breakers = { + tid("<|channel|>"), + tid("<|start|>"), + tid("<|end|>"), + tid("<|return|>"), + tid("<|call|>"), + } + # Channel names whose content is user-visible (so a dropped body matters), + # as their token sequences (e.g. ``commentary`` is two tokens). + self.visible_channel_token_seqs = { + tuple(enc.encode("final", allowed_special="all")), + tuple(enc.encode("commentary", allowed_special="all")), + } + + def is_whitespace(self, token_id: int) -> bool: + try: + return self._enc.decode([token_id]).strip() == "" + except Exception: + return False + + +_harmony_control: _HarmonyControlTokens | None = None + + +def _get_harmony_control() -> _HarmonyControlTokens: + global _harmony_control + if _harmony_control is None: + _harmony_control = _HarmonyControlTokens() + return _harmony_control + + +class _RepairingStreamableParser(StreamableParser): + """``StreamableParser`` that repairs a visible channel header emitted without + its ``<|message|>`` delimiter. + + GPT-OSS occasionally emits a user-visible channel header (``<|channel|>final`` + or ``<|channel|>commentary``) directly followed by the message body, omitting + the required ``<|message|>`` delimiter. The base parser then never leaves the + header state and the generated answer is silently dropped (the response + returns ``content=None`` even though the tokens were generated and billed). + + This subclass watches the token stream and inserts the missing ``<|message|>`` + before the first content token, so the body flows through the normal (lossless) + content path. It is a no-op on well-formed output. Because every consumer -- + streaming chat, non-streaming chat, and the Responses API -- builds its parser + through ``get_streamable_parser_for_assistant``, repairing here covers them all. + """ + + def __init__(self, encoding: Any, role: Any, *, strict: bool = True) -> None: + super().__init__(encoding, role, strict=strict) + self._ctrl = _get_harmony_control() + # State of the small header-tracking machine. + self._reading_channel_name = False # just saw ``<|channel|>`` + self._channel_name: tuple[int, ...] = () + self._awaiting_delimiter = False # in a visible header, no delimiter yet + + def process(self, token: int) -> "StreamableParser": + ctrl = self._ctrl + if self._awaiting_delimiter: + terminal = token in (ctrl.message, ctrl.constrain, ctrl.recipient) + if terminal or token in ctrl.breakers: + # Well-formed (``<|message|>``), typed, tool-call, or empty header. + self._awaiting_delimiter = False + elif not ctrl.is_whitespace(token): + # Content with no delimiter: insert ``<|message|>`` before it. + super().process(ctrl.message) + self._awaiting_delimiter = False + return super().process(token) + if self._reading_channel_name: + self._channel_name += (token,) + name = self._channel_name + seqs = ctrl.visible_channel_token_seqs + if name in seqs: + self._reading_channel_name = False + self._awaiting_delimiter = True + elif not any(seq[: len(name)] == name for seq in seqs): + self._reading_channel_name = False # analysis / unrecognized + return super().process(token) + if token == ctrl.channel: + self._reading_channel_name = True + self._channel_name = () + return super().process(token) + + def get_streamable_parser_for_assistant() -> StreamableParser: - return StreamableParser(get_encoding(), role=Role.ASSISTANT) + return _RepairingStreamableParser(get_encoding(), role=Role.ASSISTANT) def parse_output_into_messages(token_ids: Iterable[int]) -> StreamableParser: From 2aeb16bcc3f241ec14849aadee1a32e6a0261a72 Mon Sep 17 00:00:00 2001 From: ankrovv Date: Wed, 1 Jul 2026 15:49:59 -0700 Subject: [PATCH 2/3] fix(gpt-oss): enforce Harmony terminal content invariant Recover only leak-safe visible output when Harmony parsing ends empty. Mark unrecoverable Chat Completions and Responses output as explicit content_null incompletes while preserving tool-only responses in streaming and non-streaming paths. Assisted-by: OpenAI Codex Signed-off-by: ankrovv --- .../chat_completion/test_serving_chat.py | 67 +++- .../openai/parser/test_harmony_utils.py | 220 ++++++++++++ .../entrypoints/openai/responses/conftest.py | 16 +- .../openai/responses/test_harmony_utils.py | 65 ++++ .../openai/responses/test_protocol.py | 28 ++ .../responses/test_serving_responses.py | 68 +++- tests/entrypoints/openai/utils.py | 3 + .../openai/chat_completion/serving.py | 150 ++++++-- .../openai/parser/harmony_utils.py | 320 +++++++++++++++++- vllm/entrypoints/openai/responses/context.py | 81 ++++- vllm/entrypoints/openai/responses/harmony.py | 54 +++ vllm/entrypoints/openai/responses/protocol.py | 13 + vllm/entrypoints/openai/responses/serving.py | 108 ++++-- 13 files changed, 1110 insertions(+), 83 deletions(-) diff --git a/tests/entrypoints/openai/chat_completion/test_serving_chat.py b/tests/entrypoints/openai/chat_completion/test_serving_chat.py index b16ef54e4d9c..4167d0847ad5 100644 --- a/tests/entrypoints/openai/chat_completion/test_serving_chat.py +++ b/tests/entrypoints/openai/chat_completion/test_serving_chat.py @@ -1284,6 +1284,7 @@ async def generate_response_from_harmony_str( req: ChatCompletionRequest, harmony_str: str, stream: bool = False, + finish_reason: str | None = None, ) -> ChatCompletionResponse: harmony_token_ids = get_encoding().encode(harmony_str, allowed_special="all") @@ -1294,11 +1295,14 @@ async def result_generator(): req, [token_id] ) yield self.mock_request_output_from_req_and_token_ids( - req, [], finished=True + req, [], finished=True, finish_reason=finish_reason ) else: yield self.mock_request_output_from_req_and_token_ids( - req, harmony_token_ids, finished=True + req, + harmony_token_ids, + finished=True, + finish_reason=finish_reason, ) generator_func = ( @@ -1487,8 +1491,8 @@ async def test_harmony_required_tool_choice_render_request_sets_json_schema( tool_choice="required", ) - generate_request = await ( - serving_chat.openai_serving_render.render_chat_request(req) + generate_request = await serving_chat.openai_serving_render.render_chat_request( + req ) assert not isinstance(generate_request, ErrorResponse) @@ -1658,6 +1662,61 @@ async def test_simple_chat(self, serving_chat, stream): ], ) + @pytest.mark.asyncio + async def test_malformed_constraint_recovers_filtered_json( + self, serving_chat, stream + ): + req = ChatCompletionRequest( + model=MODEL_NAME, + messages=[{"role": "user", "content": "Return JSON."}], + ) + response_str = ( + "<|channel|>analysis<|message|>private reasoning<|end|>" + "<|start|>assistant<|channel|>final " + '<|constrain|>response{"ok":true}<|return|>' + ) + response = await self.generate_response_from_harmony_str( + serving_chat, + req, + response_str, + stream=stream, + finish_reason="stop", + ) + + choice = response.choices[0] + assert choice.message.content == '{"ok":true}' + assert choice.finish_reason == "stop" + assert choice.stop_reason is None + + @pytest.mark.asyncio + @pytest.mark.parametrize( + "response_str", + [ + "<|channel|>analysis<|message|>private reasoning<|return|>", + '<|channel|>final_output<|message|>{"ambiguous":true}<|return|>', + ], + ids=["analysis_only", "unknown_channel"], + ) + async def test_empty_or_ambiguous_output_is_explicitly_incomplete( + self, serving_chat, stream, response_str + ): + req = ChatCompletionRequest( + model=MODEL_NAME, + messages=[{"role": "user", "content": "Return JSON."}], + ) + response = await self.generate_response_from_harmony_str( + serving_chat, + req, + response_str, + stream=stream, + finish_reason="stop", + ) + + choice = response.choices[0] + assert not (choice.message.content or "").strip() + assert choice.finish_reason == "length" + assert choice.stop_reason == "content_null" + @pytest.mark.asyncio async def test_system_message_without_tools(self, serving_chat, stream): """Leading system message produces a developer message with diff --git a/tests/entrypoints/openai/parser/test_harmony_utils.py b/tests/entrypoints/openai/parser/test_harmony_utils.py index ff6cb0bce530..7fd57a5e899e 100644 --- a/tests/entrypoints/openai/parser/test_harmony_utils.py +++ b/tests/entrypoints/openai/parser/test_harmony_utils.py @@ -8,8 +8,10 @@ from tests.entrypoints.openai.utils import verify_harmony_messages from vllm.entrypoints.openai.chat_completion.protocol import ChatCompletionToolsParam from vllm.entrypoints.openai.parser.harmony_utils import ( + HarmonyTerminalState, auto_drop_analysis_messages, build_harmony_preamble, + classify_harmony_terminal, create_tool_definition, extract_function_from_recipient, get_encoding, @@ -17,9 +19,11 @@ get_system_message, has_custom_tools, is_function_recipient, + is_visible_content_empty, parse_chat_input_to_harmony_message, parse_chat_output, parse_output_into_messages, + recover_harmony_visible_content, ) from vllm.entrypoints.openai.responses.harmony import ( response_input_to_harmony, @@ -1312,6 +1316,30 @@ def test_well_formed_final_is_noop(self): assert content == '{"answer": "hi"}' assert _streamed_final(tokens) == '{"answer": "hi"}' + def test_header_junk_before_real_delimiter_is_not_content(self): + tokens = [ + *_analysis("thinking"), + *_assistant_header(), + _CHANNEL, + *_tok("final JSON"), + _MESSAGE, + *_tok('{"answer": "hi"}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content == '{"answer": "hi"}' + assert _streamed_final(tokens) == '{"answer": "hi"}' + + def test_missing_delimiter_rejects_identifier_metadata(self): + tokens = [ + _CHANNEL, + *_tok("final stray-metadata"), + *_tok('{"answer": "hi"}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content is None + def test_analysis_only_has_no_visible_content(self): tokens = [_CHANNEL, *_tok("analysis"), _MESSAGE, *_tok("reasoning"), _RETURN] _, content, _ = parse_chat_output(tokens) @@ -1337,3 +1365,195 @@ def test_tool_call_commentary_is_not_treated_as_content(self): ] _, content, _ = parse_chat_output(tokens) assert content is None + + def test_unknown_channel_without_delimiter_is_not_repaired(self): + tokens = [ + _CHANNEL, + *_tok("final_output"), + *_tok('{"ambiguous": true}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content is None + + @pytest.mark.parametrize("suffix", ["_output", ".output", " output"]) + def test_unknown_channel_continuations_are_not_repaired(self, suffix): + tokens = [ + _CHANNEL, + *_tok(f"final{suffix}"), + *_tok('{"ambiguous": true}'), + _RETURN, + ] + _, content, _ = parse_chat_output(tokens) + assert content is None + + +class TestHarmonyTerminalClassification: + def test_whitespace_is_empty(self): + assert is_visible_content_empty(None) + assert is_visible_content_empty(" \n\t") + assert not is_visible_content_empty(" answer ") + + def test_tool_calls_take_precedence_over_empty_content(self): + result = classify_harmony_terminal( + content=None, + has_tool_calls=True, + token_ids=_analysis("private reasoning"), + ) + assert result.state is HarmonyTerminalState.TOOL_CALLS + + def test_existing_content_is_unchanged(self): + result = classify_harmony_terminal( + content="answer", + has_tool_calls=False, + token_ids=[], + ) + assert result.state is HarmonyTerminalState.CONTENT + assert result.content == "answer" + + def test_recovers_exact_final_body_after_constraint(self): + tokens = [ + _CHANNEL, + *_tok("final"), + *_tok("<|constrain|>"), + *_tok("json"), + _MESSAGE, + *_tok('{"answer": "hi"}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) == '{"answer": "hi"}' + result = classify_harmony_terminal( + content=None, + has_tool_calls=False, + token_ids=tokens, + ) + assert result.state is HarmonyTerminalState.RECOVERED_CONTENT + assert result.content == '{"answer": "hi"}' + + def test_recovers_valid_json_after_malformed_constraint_metadata(self): + tokens = [ + _CHANNEL, + *_tok("final"), + *_tok(" "), + *_tok("<|constrain|>"), + *_tok('response{"answer": "hi"}'), + _RETURN, + ] + _, parsed_content, _ = parse_chat_output(tokens) + assert parsed_content is None + assert recover_harmony_visible_content(tokens) == '{"answer": "hi"}' + + def test_does_not_recover_non_json_after_constraint_without_delimiter(self): + tokens = [ + _CHANNEL, + *_tok("final"), + *_tok("<|constrain|>"), + *_tok("formatting private or ambiguous text"), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) is None + + def test_recovers_exact_final_with_missing_delimiter(self): + tokens = [ + _CHANNEL, + *_tok("final"), + *_tok('{"answer": "hi"}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) == '{"answer": "hi"}' + + def test_strips_punctuation_around_json_with_missing_delimiter(self): + tokens = [ + _CHANNEL, + *_tok("commentary"), + *_tok('(|{"answer": "hi"})'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) == '{"answer": "hi"}' + + def test_strips_non_ascii_artifact_after_whitespace(self): + tokens = [ + _CHANNEL, + *_tok("final 日日"), + *_tok('{"answer": "hi"}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) == '{"answer": "hi"}' + + def test_recovers_recipientless_commentary(self): + tokens = [ + _CHANNEL, + *_tok("commentary"), + _MESSAGE, + *_tok("Visible preamble"), + _END, + ] + assert recover_harmony_visible_content(tokens) == "Visible preamble" + + def test_does_not_recover_analysis(self): + tokens = _analysis("private reasoning") + result = classify_harmony_terminal( + content=None, + has_tool_calls=False, + token_ids=tokens, + ) + assert result.state is HarmonyTerminalState.CONTENT_NULL + assert result.content == "" + + def test_does_not_recover_unknown_final_like_channel(self): + tokens = [ + _CHANNEL, + *_tok("final_output"), + _MESSAGE, + *_tok("ambiguous body"), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) is None + + @pytest.mark.parametrize("suffix", ["_output", ".output", " output"]) + def test_does_not_recover_unknown_channel_without_delimiter(self, suffix): + tokens = [ + _CHANNEL, + *_tok(f"final{suffix}"), + *_tok('{"ambiguous": true}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) is None + + def test_does_not_recover_tool_recipient(self): + tokens = [ + _START, + *_tok("assistant to=functions.get_weather"), + _CHANNEL, + *_tok("commentary"), + _MESSAGE, + *_tok('{"city": "Paris"}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) is None + + def test_does_not_recover_recipient_after_channel(self): + tokens = [ + _CHANNEL, + *_tok("commentary"), + *_tok(" to=functions.get_weather"), + *_tok('{"city": "Paris"}'), + _RETURN, + ] + assert recover_harmony_visible_content(tokens) is None + + def test_recovery_can_be_disabled(self): + tokens = [ + _CHANNEL, + *_tok("final"), + _MESSAGE, + *_tok("raw required-tool JSON"), + _RETURN, + ] + result = classify_harmony_terminal( + content=None, + has_tool_calls=False, + token_ids=tokens, + allow_recovery=False, + ) + assert result.state is HarmonyTerminalState.CONTENT_NULL diff --git a/tests/entrypoints/openai/responses/conftest.py b/tests/entrypoints/openai/responses/conftest.py index a1d16b123166..84bd7f68be65 100644 --- a/tests/entrypoints/openai/responses/conftest.py +++ b/tests/entrypoints/openai/responses/conftest.py @@ -33,6 +33,7 @@ def pairs_of_event_types() -> dict[str, str]: # fmt: off event_pairs = { "response.completed": "response.created", + "response.incomplete": "response.created", "response.output_item.done": "response.output_item.added", "response.content_part.done": "response.content_part.added", "response.output_text.done": "response.output_text.delta", @@ -150,9 +151,9 @@ def _validate_event_ordering(events: list) -> None: assert events[0].type == "response.created", ( f"First event must be response.created, got {events[0].type}" ) - # Last event must be response.completed - assert events[-1].type == "response.completed", ( - f"Last event must be response.completed, got {events[-1].type}" + # Last event must be a terminal response envelope. + assert events[-1].type in {"response.completed", "response.incomplete"}, ( + f"Last event must be terminal, got {events[-1].type}" ) # response.in_progress, if present, must be the second event @@ -165,14 +166,16 @@ def _validate_event_ordering(events: list) -> None: f"found at indices {in_progress_indices}" ) - # Exactly one created and one completed + # Exactly one created and one terminal event. created_count = sum(1 for e in events if e.type == "response.created") completed_count = sum(1 for e in events if e.type == "response.completed") + incomplete_count = sum(1 for e in events if e.type == "response.incomplete") assert created_count == 1, ( f"Expected exactly 1 response.created, got {created_count}" ) - assert completed_count == 1, ( - f"Expected exactly 1 response.completed, got {completed_count}" + assert completed_count + incomplete_count == 1, ( + "Expected exactly 1 terminal response event, got " + f"completed={completed_count}, incomplete={incomplete_count}" ) @@ -187,6 +190,7 @@ def _validate_field_consistency(events: list) -> None: "response.created", "response.in_progress", "response.completed", + "response.incomplete", } active_item_id: str | None = None diff --git a/tests/entrypoints/openai/responses/test_harmony_utils.py b/tests/entrypoints/openai/responses/test_harmony_utils.py index f1434ce2bd58..da244a900873 100644 --- a/tests/entrypoints/openai/responses/test_harmony_utils.py +++ b/tests/entrypoints/openai/responses/test_harmony_utils.py @@ -10,7 +10,12 @@ from openai.types.responses.response_output_item import McpCall from openai_harmony import Author, Message, Role, TextContent +from vllm.entrypoints.openai.parser.harmony_utils import ( + HarmonyTerminalState, + get_encoding, +) from vllm.entrypoints.openai.responses.harmony import ( + apply_harmony_terminal_invariant, harmony_to_response_output, parser_state_to_response_output, response_previous_input_to_harmony, @@ -580,6 +585,66 @@ def test_parser_state_to_response_output_commentary_channel() -> None: assert preamble_items[0].status == "incomplete" # streaming +class TestHarmonyResponsesTerminalInvariant: + @staticmethod + def _tokens(text: str) -> list[int]: + return get_encoding().encode(text, allowed_special="all") + + def test_recovers_filtered_final_json(self) -> None: + token_ids = self._tokens( + '<|channel|>final<|constrain|>formatting{"ok":true}<|return|>' + ) + + output, terminal = apply_harmony_terminal_invariant([], token_ids) + + assert terminal.state is HarmonyTerminalState.RECOVERED_CONTENT + assert len(output) == 1 + assert isinstance(output[0], ResponseOutputMessage) + assert output[0].content[0].text == '{"ok":true}' + + def test_ambiguous_channel_fails_closed(self) -> None: + token_ids = self._tokens( + '<|channel|>final_output<|message|>{"unsafe":true}<|return|>' + ) + + output, terminal = apply_harmony_terminal_invariant([], token_ids) + + assert terminal.state is HarmonyTerminalState.CONTENT_NULL + assert output == [] + + def test_tool_call_takes_precedence(self) -> None: + tool_call = ResponseFunctionToolCall( + arguments='{"city":"Paris"}', + call_id="call_test", + name="get_weather", + type="function_call", + id="fc_test", + status="completed", + ) + + output, terminal = apply_harmony_terminal_invariant([tool_call], []) + + assert terminal.state is HarmonyTerminalState.TOOL_CALLS + assert output == [tool_call] + + def test_analysis_only_fails_closed(self) -> None: + reasoning = ResponseReasoningItem( + id="rs_test", + summary=[], + type="reasoning", + content=[], + status=None, + ) + token_ids = self._tokens( + "<|channel|>analysis<|message|>hidden reasoning<|return|>" + ) + + output, terminal = apply_harmony_terminal_invariant([reasoning], token_ids) + + assert terminal.state is HarmonyTerminalState.CONTENT_NULL + assert output == [reasoning] + + def test_parser_state_to_response_output_analysis_channel() -> None: """Test parser_state_to_response_output with analysis channel and various recipients.""" diff --git a/tests/entrypoints/openai/responses/test_protocol.py b/tests/entrypoints/openai/responses/test_protocol.py index db5d7d692490..97d4660b4aff 100644 --- a/tests/entrypoints/openai/responses/test_protocol.py +++ b/tests/entrypoints/openai/responses/test_protocol.py @@ -5,9 +5,13 @@ ) from vllm.entrypoints.openai.responses.protocol import ( + ResponseIncompleteEvent, + ResponsesRequest, + ResponsesResponse, serialize_message, serialize_messages, ) +from vllm.sampling_params import SamplingParams def test_serialize_message() -> None: @@ -37,3 +41,27 @@ def test_serialize_messages() -> None: } msg = Message.from_dict(msg_value) assert serialize_messages([msg, dict_value]) == [msg_value, dict_value] + + +def test_content_null_incomplete_response_contract() -> None: + request = ResponsesRequest(input="hello") + response = ResponsesResponse.from_request( + request, + SamplingParams(max_tokens=16), + model_name="test-model", + created_time=123, + output=[], + status="incomplete", + stop_reason="content_null", + ) + + event = ResponseIncompleteEvent( + type="response.incomplete", + sequence_number=2, + response=response, + ) + dumped = event.model_dump(mode="json") + + assert dumped["response"]["status"] == "incomplete" + assert dumped["response"]["stop_reason"] == "content_null" + assert dumped["response"]["incomplete_details"] == {"reason": "max_output_tokens"} diff --git a/tests/entrypoints/openai/responses/test_serving_responses.py b/tests/entrypoints/openai/responses/test_serving_responses.py index 90484250a2d6..6726517c00ba 100644 --- a/tests/entrypoints/openai/responses/test_serving_responses.py +++ b/tests/entrypoints/openai/responses/test_serving_responses.py @@ -25,7 +25,7 @@ Mcp, Tool, ) -from openai_harmony import Role +from openai_harmony import HarmonyError, Role import vllm.envs as envs from vllm.entrypoints.mcp.tool_server import ToolServer @@ -40,7 +40,12 @@ get_encoding, render_for_completion, ) -from vllm.entrypoints.openai.responses.context import ConversationContext, SimpleContext +from vllm.entrypoints.openai.responses.context import ( + ConversationContext, + HarmonyContext, + SimpleContext, + StreamingHarmonyContext, +) from vllm.entrypoints.openai.responses.protocol import ( ResponseCreatedEvent, ResponseRawMessageAndToken, @@ -920,6 +925,65 @@ def test_commentary_with_recipient_no_preamble_done(self) -> None: assert "response.output_text.done" not in type_names +def _make_harmony_request_output( + token_ids: list[int], *, finished: bool = True +) -> RequestOutput: + completion = CompletionOutput( + index=0, + text="", + token_ids=token_ids, + cumulative_logprob=0.0, + logprobs=None, + finish_reason="stop" if finished else None, + stop_reason=None, + ) + return RequestOutput( + request_id="req_harmony_terminal", + prompt="hi", + prompt_token_ids=[7, 8], + prompt_logprobs=None, + outputs=[completion], + finished=finished, + num_cached_tokens=0, + ) + + +def test_harmony_context_defers_parse_error_to_terminal(monkeypatch) -> None: + from vllm.entrypoints.openai.responses import context as context_module + + context = HarmonyContext([], []) + failing_parser = MagicMock() + failing_parser.process.side_effect = HarmonyError("malformed header") + reset_parser = MagicMock() + parsers = iter((failing_parser, reset_parser)) + monkeypatch.setattr( + context_module, + "get_streamable_parser_for_assistant", + lambda: next(parsers), + ) + + context.append_output(_make_harmony_request_output([101, 102])) + + assert context.harmony_parse_failed is True + assert context.output_token_ids == [101, 102] + assert context.finish_reason == "stop" + assert context.messages == [] + + +def test_streaming_harmony_context_defers_parse_error_to_terminal() -> None: + context = StreamingHarmonyContext([], []) + context.parser = MagicMock() + context.parser.process.side_effect = HarmonyError("malformed header") + + context.append_output(_make_harmony_request_output([201, 202])) + + assert context.harmony_parse_failed is True + assert context.output_token_ids == [201, 202] + assert context.finish_reason == "stop" + assert context.last_output_finished is True + assert context.last_content_delta is None + + def _make_simple_context_with_output(text, token_ids): """Create a SimpleContext with a RequestOutput containing the given text.""" ctx = SimpleContext() diff --git a/tests/entrypoints/openai/utils.py b/tests/entrypoints/openai/utils.py index 36056a44d079..54b5d8fce280 100644 --- a/tests/entrypoints/openai/utils.py +++ b/tests/entrypoints/openai/utils.py @@ -27,6 +27,7 @@ async def accumulate_streaming_response( accumulated_tool_calls: list[dict[str, Any]] = [] role = None finish_reason = None + stop_reason = None response_id = None created = None model = None @@ -90,6 +91,7 @@ async def accumulate_streaming_response( if choice.finish_reason: finish_reason = choice.finish_reason + stop_reason = choice.stop_reason if choice.index is not None: index = choice.index @@ -117,6 +119,7 @@ async def accumulate_streaming_response( index=index, message=message, finish_reason=finish_reason or "stop", + stop_reason=stop_reason, ) # Create usage info (with dummy values for tests) diff --git a/vllm/entrypoints/openai/chat_completion/serving.py b/vllm/entrypoints/openai/chat_completion/serving.py index 58a90213ae9c..fa0a2ce54e93 100644 --- a/vllm/entrypoints/openai/chat_completion/serving.py +++ b/vllm/entrypoints/openai/chat_completion/serving.py @@ -13,6 +13,7 @@ import numpy as np import pybase64 as base64 from fastapi import Request +from openai_harmony import HarmonyError from vllm.engine.protocol import EngineClient from vllm.entrypoints.chat_utils import ( @@ -56,6 +57,9 @@ ) from vllm.entrypoints.openai.models.serving import OpenAIServingModels from vllm.entrypoints.openai.parser.harmony_utils import ( + CONTENT_NULL_STOP_REASON, + HarmonyTerminalState, + classify_harmony_terminal, get_streamable_parser_for_assistant, parse_chat_output, ) @@ -421,8 +425,10 @@ def _is_harmony_required_tool_choice( self, request: ChatCompletionRequest, ) -> bool: - return self.use_harmony and request.tool_choice == "required" and bool( - request.tools + return ( + self.use_harmony + and request.tool_choice == "required" + and bool(request.tools) ) def _parse_harmony_required_tool_calls( @@ -500,14 +506,18 @@ async def chat_completion_stream_generator( finish_reason_sent = [False] * num_choices num_prompt_tokens = 0 num_cached_tokens = None - is_harmony_required_tool_choice = self._is_harmony_required_tool_choice( - request - ) + is_harmony_required_tool_choice = self._is_harmony_required_tool_choice(request) + harmony_output_token_ids: list[list[int]] | None = None + harmony_visible_content: list[str] | None = None + harmony_parse_failed: list[bool] | None = None if self.use_harmony: harmony_parsers = [ get_streamable_parser_for_assistant() for _ in range(num_choices) ] harmony_tools_streamed = [False] * num_choices + harmony_output_token_ids = [[] for _ in range(num_choices)] + harmony_visible_content = [""] * num_choices + harmony_parse_failed = [False] * num_choices required_tool_name_returned = [False] * num_choices tools_streamed = [False] * num_choices @@ -680,6 +690,10 @@ async def chat_completion_stream_generator( parser = parsers[i] tool_parser = parser.tool_parser if parser is not None else None + if self.use_harmony: + assert harmony_output_token_ids is not None + harmony_output_token_ids[i].extend(output.token_ids) + if ( reasoning_parser and res.prompt_token_ids @@ -709,22 +723,44 @@ async def chat_completion_stream_generator( delta_text = output.text elif self.use_harmony: harmony_parser = harmony_parsers[i] - prev_recipient = harmony_parser.current_recipient + assert harmony_parse_failed is not None + prev_recipient = ( + None + if harmony_parse_failed[i] + else harmony_parser.current_recipient + ) # Track accumulated content per token with their state token_states: list[TokenState] = [] for token_id in output.token_ids: - harmony_parser.process(token_id) - token_delta = harmony_parser.last_content_delta or "" - token_states.append( - TokenState( - harmony_parser.current_channel, - harmony_parser.current_recipient, - token_delta, + if harmony_parse_failed[i]: + token_states.append(TokenState(None, None, "")) + continue + try: + harmony_parser.process(token_id) + token_delta = harmony_parser.last_content_delta or "" + token_states.append( + TokenState( + harmony_parser.current_channel, + harmony_parser.current_recipient, + token_delta, + ) + ) + except HarmonyError: + harmony_parse_failed[i] = True + token_states.append(TokenState(None, None, "")) + logger.warning( + "Harmony streaming parse failed for request %s " + "choice %d; deferring to terminal recovery.", + request_id, + i, ) - ) delta_text = "".join(delta for _, _, delta in token_states) - cur_channel = harmony_parser.current_channel + cur_channel = ( + None + if harmony_parse_failed[i] + else harmony_parser.current_channel + ) # handle the case where several tokens where generated at once # including the final token, leading to a delta in the text @@ -844,6 +880,10 @@ async def chat_completion_stream_generator( else: delta_message = DeltaMessage(content=delta_text) + if self.use_harmony and delta_message and delta_message.content: + assert harmony_visible_content is not None + harmony_visible_content[i] += delta_message.content + # update the previous values for the next iteration if ( is_mistral_grammar_path @@ -945,12 +985,45 @@ async def chat_completion_stream_generator( finish_reason_ = ( output.finish_reason if output.finish_reason else "stop" ) + stop_reason_ = output.stop_reason + if self.use_harmony: + assert harmony_output_token_ids is not None + assert harmony_visible_content is not None + terminal = classify_harmony_terminal( + content=harmony_visible_content[i], + has_tool_calls=( + tools_streamed[i] or harmony_tools_streamed[i] + ), + token_ids=harmony_output_token_ids[i], + allow_recovery=not is_harmony_required_tool_choice, + ) + if terminal.state is HarmonyTerminalState.RECOVERED_CONTENT: + delta_message.content = terminal.content + logger.warning( + "Recovered filtered Harmony content for streaming " + "request %s choice %d.", + request_id, + i, + ) + elif terminal.state is HarmonyTerminalState.CONTENT_NULL: + # Never terminate a no-tool Harmony choice as + # 200/stop with empty visible content. + delta_message.content = "" + finish_reason_ = "length" + stop_reason_ = CONTENT_NULL_STOP_REASON + logger.warning( + "Harmony produced no recoverable visible content " + "for streaming request %s choice %d; marking it " + "content_null.", + request_id, + i, + ) choice_data = ChatCompletionResponseStreamChoice( index=i, delta=delta_message, logprobs=logprobs, finish_reason=finish_reason_, - stop_reason=output.stop_reason, + stop_reason=stop_reason_, token_ids=( as_list(output.token_ids) if request.return_token_ids @@ -1192,18 +1265,47 @@ async def chat_completion_full_generator( "ascii" ) + finish_reason_ = ( + "tool_calls" + if (tool_call_info is not None and tool_call_info.tools_called) + else output.finish_reason + if output.finish_reason + else "stop" + ) + stop_reason_ = output.stop_reason + terminal = classify_harmony_terminal( + content=( + message.content if isinstance(message.content, str) else None + ), + has_tool_calls=bool(message.tool_calls), + token_ids=token_ids, + allow_recovery=not self._is_harmony_required_tool_choice(request), + ) + if terminal.state is HarmonyTerminalState.RECOVERED_CONTENT: + message.content = terminal.content + logger.warning( + "Recovered filtered Harmony content for request %s choice %d.", + request_id, + output.index, + ) + elif terminal.state is HarmonyTerminalState.CONTENT_NULL: + # Override the engine stop locally; do not mutate RequestOutput. + message.content = "" + finish_reason_ = "length" + stop_reason_ = CONTENT_NULL_STOP_REASON + logger.warning( + "Harmony produced no recoverable visible content for request " + "%s choice %d; marking it content_null.", + request_id, + output.index, + ) + choice_data = ChatCompletionResponseChoice( index=output.index, message=message, logprobs=logprobs, - finish_reason=( - "tool_calls" - if (tool_call_info is not None and tool_call_info.tools_called) - else output.finish_reason - if output.finish_reason - else "stop" - ), - stop_reason=output.stop_reason, + finish_reason=finish_reason_, + stop_reason=stop_reason_, token_ids=( as_list(output.token_ids) if request.return_token_ids else None ), diff --git a/vllm/entrypoints/openai/parser/harmony_utils.py b/vllm/entrypoints/openai/parser/harmony_utils.py index d4f111fa7227..d23740edb9bd 100644 --- a/vllm/entrypoints/openai/parser/harmony_utils.py +++ b/vllm/entrypoints/openai/parser/harmony_utils.py @@ -2,7 +2,10 @@ # SPDX-FileCopyrightText: Copyright contributors to the vLLM project import datetime +import json from collections.abc import Iterable, Sequence +from dataclasses import dataclass +from enum import Enum from typing import Any from openai.types.responses.tool import Tool @@ -11,6 +14,7 @@ Conversation, DeveloperContent, HarmonyEncodingName, + HarmonyError, Message, ReasoningEffort, Role, @@ -27,6 +31,29 @@ logger = init_logger(__name__) +CONTENT_NULL_STOP_REASON = "content_null" + + +class HarmonyTerminalState(Enum): + """Outcome of classifying one completed Harmony choice.""" + + TOOL_CALLS = "tool_calls" + CONTENT = "content" + RECOVERED_CONTENT = "recovered_content" + CONTENT_NULL = "content_null" + + +@dataclass(frozen=True) +class HarmonyTerminalResult: + state: HarmonyTerminalState + content: str | None = None + + +def is_visible_content_empty(content: str | None) -> bool: + """Return whether content is absent or only whitespace.""" + + return content is None or not content.strip() + def is_function_recipient( recipient: str, @@ -479,13 +506,17 @@ def tid(text: str) -> int: self.message = tid("<|message|>") self.constrain = tid("<|constrain|>") self.recipient = tid(" to") # start of a ` to=` tool target + self.start = tid("<|start|>") + self.end = tid("<|end|>") + self.return_ = tid("<|return|>") + self.call = tid("<|call|>") # Tokens that terminate or restart a header; content never starts with one. self.breakers = { tid("<|channel|>"), - tid("<|start|>"), - tid("<|end|>"), - tid("<|return|>"), - tid("<|call|>"), + self.start, + self.end, + self.return_, + self.call, } # Channel names whose content is user-visible (so a dropped body matters), # as their token sequences (e.g. ``commentary`` is two tokens). @@ -511,6 +542,192 @@ def _get_harmony_control() -> _HarmonyControlTokens: return _harmony_control +def _matches_at( + token_ids: Sequence[int], start: int, expected: tuple[int, ...] +) -> bool: + end = start + len(expected) + return end <= len(token_ids) and tuple(token_ids[start:end]) == expected + + +def _header_has_recipient( + token_ids: Sequence[int], channel_index: int, message_index: int +) -> bool: + """Conservatively detect a tool recipient around a channel header.""" + + ctrl = _get_harmony_control() + header_start = 0 + for index in range(channel_index - 1, -1, -1): + if token_ids[index] in ctrl.breakers: + header_start = index + 1 + break + return ctrl.recipient in token_ids[header_start:message_index] + + +def _extract_complete_json(text: str, *, allow_metadata_prefix: bool) -> str | None: + """Return one complete JSON object/array surrounded only by allowed metadata.""" + + decoder = json.JSONDecoder() + harmless_wrapper_chars = frozenset(" \t\r\n()|`") + for index, char in enumerate(text): + if char not in "{[": + continue + try: + _, end = decoder.raw_decode(text[index:]) + except json.JSONDecodeError: + continue + prefix = text[:index] + suffix = text[index + end :] + if any(char not in harmless_wrapper_chars for char in suffix): + continue + if not allow_metadata_prefix and any( + char not in harmless_wrapper_chars for char in prefix + ): + continue + return text[index : index + end] + return None + + +def _extract_json_from_missing_body(text: str) -> str | None: + """Recover JSON while rejecting identifier-like channel continuations.""" + + recovered = _extract_complete_json(text, allow_metadata_prefix=True) + if recovered is None: + return None + prefix = text[: text.find(recovered)] + harmless_wrapper_chars = frozenset(' \t\r\n()|`".') + if all(char in harmless_wrapper_chars for char in prefix): + return recovered + if prefix[:1].isspace() and not any( + char.isascii() and (char.isalnum() or char == "_") for char in prefix + ): + return recovered + return None + + +def _is_plausible_missing_message_body(text: str) -> bool: + """Reject channel-name continuations such as ``final_output``.""" + + return bool(text) and (text[0].isspace() or text[0] in '{[(.|`"') + + +def recover_harmony_visible_content(token_ids: Sequence[int]) -> str | None: + """Recover only unambiguously user-visible Harmony message bodies. + + This is intentionally narrower than decoding the raw model output. It accepts + exact ``final`` and recipient-less ``commentary`` channel headers, optionally + with a ``<|constrain|>...`` content type. It decodes only the delimited body, + or the body up to a terminal control token when ``<|message|>`` is missing. + Analysis, tool recipients, unknown channels, ambiguous nested headers, and + Harmony control tokens are never returned. + """ + + ctrl = _get_harmony_control() + recovered: list[str] = [] + + for channel_index, token_id in enumerate(token_ids): + if token_id != ctrl.channel: + continue + + for channel_tokens in ctrl.visible_channel_token_seqs: + channel_start = channel_index + 1 + if not _matches_at(token_ids, channel_start, channel_tokens): + continue + + cursor = channel_start + len(channel_tokens) + while cursor < len(token_ids) and ctrl.is_whitespace(token_ids[cursor]): + cursor += 1 + if cursor >= len(token_ids): + continue + + constrained_without_delimiter = False + if token_ids[cursor] == ctrl.constrain: + cursor += 1 + constrained_start = cursor + while cursor < len(token_ids) and token_ids[cursor] != ctrl.message: + if token_ids[cursor] in ctrl.breakers: + break + cursor += 1 + constrained_without_delimiter = ( + cursor >= len(token_ids) or token_ids[cursor] != ctrl.message + ) + + if constrained_without_delimiter: + if _header_has_recipient( + token_ids, channel_index, constrained_start + ): + continue + raw_candidate = ctrl._enc.decode( + token_ids[constrained_start:cursor] + ) + recovered_json = _extract_complete_json( + raw_candidate, allow_metadata_prefix=True + ) + if recovered_json is not None: + recovered.append(recovered_json) + break + + if cursor >= len(token_ids): + continue + if token_ids[cursor] == ctrl.recipient: + continue + has_message_delimiter = token_ids[cursor] == ctrl.message + if not has_message_delimiter and token_ids[cursor] in ctrl.breakers: + continue + if _header_has_recipient(token_ids, channel_index, cursor): + continue + + body_start = cursor + 1 if has_message_delimiter else cursor + body_end = body_start + while ( + body_end < len(token_ids) and token_ids[body_end] not in ctrl.breakers + ): + body_end += 1 + + body_tokens = token_ids[body_start:body_end] + # Nested header controls make the candidate ambiguous. Fail closed. + if ctrl.message in body_tokens or ctrl.constrain in body_tokens: + continue + + body = ctrl._enc.decode(body_tokens) + if not has_message_delimiter: + if not _is_plausible_missing_message_body(body): + continue + recovered_json = _extract_json_from_missing_body(body) + if recovered_json is not None: + body = recovered_json + elif ( + _extract_complete_json(body, allow_metadata_prefix=True) is not None + ): + continue + if not is_visible_content_empty(body): + recovered.append(body) + break + + return "\n".join(recovered) or None + + +def classify_harmony_terminal( + *, + content: str | None, + has_tool_calls: bool, + token_ids: Sequence[int], + allow_recovery: bool = True, +) -> HarmonyTerminalResult: + """Classify a terminal Harmony choice without exposing hidden channels.""" + + if has_tool_calls: + return HarmonyTerminalResult(HarmonyTerminalState.TOOL_CALLS, content) + if not is_visible_content_empty(content): + return HarmonyTerminalResult(HarmonyTerminalState.CONTENT, content) + if allow_recovery: + recovered = recover_harmony_visible_content(token_ids) + if not is_visible_content_empty(recovered): + return HarmonyTerminalResult( + HarmonyTerminalState.RECOVERED_CONTENT, recovered + ) + return HarmonyTerminalResult(HarmonyTerminalState.CONTENT_NULL, "") + + class _RepairingStreamableParser(StreamableParser): """``StreamableParser`` that repairs a visible channel header emitted without its ``<|message|>`` delimiter. @@ -521,11 +738,13 @@ class _RepairingStreamableParser(StreamableParser): header state and the generated answer is silently dropped (the response returns ``content=None`` even though the tokens were generated and billed). - This subclass watches the token stream and inserts the missing ``<|message|>`` - before the first content token, so the body flows through the normal (lossless) - content path. It is a no-op on well-formed output. Because every consumer -- - streaming chat, non-streaming chat, and the Responses API -- builds its parser - through ``get_streamable_parser_for_assistant``, repairing here covers them all. + This subclass watches the token stream and defers only a malformed visible + header until it sees either a real delimiter or the message terminator. In the + latter case it inserts the missing ``<|message|>`` and replays the buffered body + through the normal content path. It is a no-op on well-formed output. Because + every consumer -- streaming chat, non-streaming chat, and the Responses API -- + builds its parser through ``get_streamable_parser_for_assistant``, repairing here + covers them all. """ def __init__(self, encoding: Any, role: Any, *, strict: bool = True) -> None: @@ -535,19 +754,79 @@ def __init__(self, encoding: Any, role: Any, *, strict: bool = True) -> None: self._reading_channel_name = False # just saw ``<|channel|>`` self._channel_name: tuple[int, ...] = () self._awaiting_delimiter = False # in a visible header, no delimiter yet + self._pending_visible_tokens: list[int] = [] + self._awaiting_constrained_delimiter = False + self._pending_constraint_tokens: list[int] = [] def process(self, token: int) -> "StreamableParser": ctrl = self._ctrl + if self._awaiting_constrained_delimiter: + if token == ctrl.message: + # The body delimiter is unambiguous. Discard malformed content- + # type metadata and let the base parser consume the body normally. + self._pending_constraint_tokens.clear() + self._awaiting_constrained_delimiter = False + return super().process(token) + if token in ctrl.breakers: + # No body delimiter appeared. Leave the parser at the exact visible + # channel and let the terminal classifier inspect the raw tokens. + self._pending_constraint_tokens.clear() + self._awaiting_constrained_delimiter = False + return self + self._pending_constraint_tokens.append(token) + return self if self._awaiting_delimiter: - terminal = token in (ctrl.message, ctrl.constrain, ctrl.recipient) - if terminal or token in ctrl.breakers: - # Well-formed (``<|message|>``), typed, tool-call, or empty header. + if token == ctrl.message: + # A real delimiter wins. Anything buffered between the exact + # channel name and this delimiter was malformed header metadata, + # not message content. + self._pending_visible_tokens.clear() self._awaiting_delimiter = False - elif not ctrl.is_whitespace(token): - # Content with no delimiter: insert ``<|message|>`` before it. - super().process(ctrl.message) + return super().process(token) + if token == ctrl.constrain: + self._pending_visible_tokens.clear() self._awaiting_delimiter = False - return super().process(token) + self._awaiting_constrained_delimiter = True + self._pending_constraint_tokens = [] + return self + if token == ctrl.recipient: + self._pending_visible_tokens.clear() + self._awaiting_delimiter = False + return super().process(token) + if token in ctrl.breakers: + pending = self._pending_visible_tokens + self._pending_visible_tokens = [] + self._awaiting_delimiter = False + if any(not ctrl.is_whitespace(item) for item in pending): + pending_text = ctrl._enc.decode(pending) + if not _is_plausible_missing_message_body(pending_text): + return super().process(token) + recovered_json = _extract_json_from_missing_body(pending_text) + if recovered_json is not None: + try: + pending = ctrl._enc.encode(recovered_json) + except ValueError: + # A literal Harmony control marker inside the JSON is + # ambiguous. Leave no content for the terminal guard. + pending = [] + elif ( + _extract_complete_json(pending_text, allow_metadata_prefix=True) + is not None + ): + return super().process(token) + # No real delimiter appeared before the message ended. The + # buffered tokens are the body, so replay them through the + # normal content state after injecting ``<|message|>``. + if pending: + super().process(ctrl.message) + for item in pending: + super().process(item) + return super().process(token) + # Delay only a malformed visible header. This lookahead prevents a + # header such as ``final JSON<|message|>...`` from being mistaken for + # content while leaving well-formed output fully streaming. + self._pending_visible_tokens.append(token) + return self if self._reading_channel_name: self._channel_name += (token,) name = self._channel_name @@ -555,6 +834,7 @@ def process(self, token: int) -> "StreamableParser": if name in seqs: self._reading_channel_name = False self._awaiting_delimiter = True + self._pending_visible_tokens = [] elif not any(seq[: len(name)] == name for seq in seqs): self._reading_channel_name = False # analysis / unrecognized return super().process(token) @@ -571,7 +851,13 @@ def get_streamable_parser_for_assistant() -> StreamableParser: def parse_output_into_messages(token_ids: Iterable[int]) -> StreamableParser: parser = get_streamable_parser_for_assistant() for token_id in token_ids: - parser.process(token_id) + try: + parser.process(token_id) + except HarmonyError: + logger.warning( + "Harmony parsing failed; deferring to filtered terminal recovery." + ) + return get_streamable_parser_for_assistant() return parser diff --git a/vllm/entrypoints/openai/responses/context.py b/vllm/entrypoints/openai/responses/context.py index e72032c24aa4..e41b7d0ac752 100644 --- a/vllm/entrypoints/openai/responses/context.py +++ b/vllm/entrypoints/openai/responses/context.py @@ -14,7 +14,7 @@ ResponseFunctionToolCallOutputItem, ) from openai.types.responses.tool import Mcp -from openai_harmony import Author, Message, Role, StreamState, TextContent +from openai_harmony import Author, HarmonyError, Message, Role, StreamState, TextContent from vllm import envs from vllm.entrypoints.chat_utils import ( @@ -546,6 +546,10 @@ def __init__( self.is_first_turn = True self.first_tok_of_message = True # For streaming support self.kv_transfer_params: dict[str, Any] | None = None + self.output_token_ids: list[int] = [] + self.harmony_parse_failed = False + self.last_output_finished = False + self._terminal_recovered_content: str | None = None def _update_num_reasoning_tokens(self): channel = self.parser.current_channel @@ -558,9 +562,23 @@ def _update_num_reasoning_tokens(self): def append_output(self, output: RequestOutput) -> None: output_token_ids = output.outputs[0].token_ids + self.output_token_ids = list(output_token_ids) + self.harmony_parse_failed = False + self.last_output_finished = output.finished + self._terminal_recovered_content = None self.parser = get_streamable_parser_for_assistant() for token_id in output_token_ids: - self.parser.process(token_id) + try: + self.parser.process(token_id) + except HarmonyError: + self.harmony_parse_failed = True + self.parser = get_streamable_parser_for_assistant() + logger.warning( + "Harmony Responses parsing failed for request %s; " + "deferring to terminal recovery.", + output.request_id, + ) + break # Check if the current token is part of reasoning content self._update_num_reasoning_tokens() self._update_prefill_token_usage(output) @@ -573,11 +591,23 @@ def append_output(self, output: RequestOutput) -> None: # append_output is called only once before tool calling # in non-streaming case # so we can append all the parser messages to _messages - output_msgs = self.parser.messages + output_msgs = [] if self.harmony_parse_failed else self.parser.messages # The responses finish reason is set in the last message self.finish_reason = output.outputs[0].finish_reason self._messages.extend(output_msgs) + def record_terminal_recovery(self, content: str) -> None: + """Persist one filtered recovery for response output and replay.""" + + if self._terminal_recovered_content == content: + return + recovered_message = Message.from_role_and_content( + Role.ASSISTANT, content + ).with_channel("final") + self._messages.append(recovered_message) + self._terminal_recovered_content = content + self.parser = get_streamable_parser_for_assistant() + def append_tool_output(self, output: list[Message]) -> None: output_msgs = output self._messages.extend(output_msgs) @@ -864,6 +894,10 @@ def append_output(self, output: RequestOutput) -> None: # append_output is called for each output token in streaming case, # so we only want to add the prompt tokens once for each message. self.last_content_delta = None + if self.first_tok_of_message: + self.output_token_ids = [] + self.harmony_parse_failed = False + self._terminal_recovered_content = None if self.first_tok_of_message: self._update_prefill_token_usage(output) # Reset self.first_tok_of_message if needed: @@ -871,24 +905,45 @@ def append_output(self, output: RequestOutput) -> None: # (finished=True), then the next token processed will mark the # beginning of a new message self.first_tok_of_message = output.finished + completion_output = output.outputs[0] + self.output_token_ids.extend(completion_output.token_ids) last_delta_text = "" - for tok in output.outputs[0].token_ids: - self.parser.process(tok) + for tok in completion_output.token_ids: + if self.harmony_parse_failed: + continue + try: + self.parser.process(tok) + except HarmonyError: + self.harmony_parse_failed = True + self.last_content_delta = None + logger.warning( + "Harmony Responses streaming parse failed for request %s; " + "deferring to terminal recovery.", + output.request_id, + ) + continue last_delta_text += self.parser.last_content_delta or "" if last_delta_text: self.last_content_delta = last_delta_text self._update_decode_token_usage(output) if output.kv_transfer_params is not None: self.kv_transfer_params = output.kv_transfer_params + self.finish_reason = completion_output.finish_reason + self.last_output = output + self.last_output_finished = output.finished # For streaming, update previous turn when message is complete if output.finished: self.all_turn_metrics.append(self.current_turn_metrics.copy()) self.current_turn_metrics.reset() # Check if the current token is part of reasoning content - self._update_num_reasoning_tokens() - self.last_tok = tok - if len(self._messages) - self.num_init_messages < len(self.parser.messages): + if not self.harmony_parse_failed: + self._update_num_reasoning_tokens() + if completion_output.token_ids: + self.last_tok = completion_output.token_ids[-1] + if not self.harmony_parse_failed and len( + self._messages + ) - self.num_init_messages < len(self.parser.messages): self._messages.extend( self.parser.messages[len(self._messages) - self.num_init_messages :] ) @@ -908,10 +963,16 @@ def append_tool_output(self, output: list[Message]) -> None: # TODO: add tool_output messages to self._messages def is_expecting_start(self) -> bool: - return self.parser.state == StreamState.EXPECT_START + return ( + not self.harmony_parse_failed + and self.parser.state == StreamState.EXPECT_START + ) def is_assistant_action_turn(self) -> bool: - return self.last_tok in self.encoding.stop_tokens_for_assistant_actions() + return ( + not self.harmony_parse_failed + and self.last_tok in self.encoding.stop_tokens_for_assistant_actions() + ) def render_for_completion(self) -> list[int]: # now this list of tokens as next turn's starting tokens diff --git a/vllm/entrypoints/openai/responses/harmony.py b/vllm/entrypoints/openai/responses/harmony.py index 8dee0d993d5f..c96733cd1874 100644 --- a/vllm/entrypoints/openai/responses/harmony.py +++ b/vllm/entrypoints/openai/responses/harmony.py @@ -9,6 +9,7 @@ """ import json +from collections.abc import Sequence from openai.types.responses import ( ResponseFunctionToolCall, @@ -31,6 +32,9 @@ from vllm.entrypoints.openai.parser.harmony_utils import ( BUILTIN_TOOL_TO_MCP_SERVER_LABEL, + HarmonyTerminalResult, + HarmonyTerminalState, + classify_harmony_terminal, extract_function_from_recipient, flatten_input_text_content, get_system_or_developer_message, @@ -578,3 +582,53 @@ def parser_state_to_response_output( return [text_item] return [] + + +def apply_harmony_terminal_invariant( + output_items: list[ResponseOutputItem], + token_ids: Sequence[int], +) -> tuple[list[ResponseOutputItem], HarmonyTerminalResult]: + """Apply the shared Harmony terminal invariant to Responses output.""" + + visible_text: list[str] = [] + has_tool_calls = False + for item in output_items: + if isinstance(item, ResponseOutputMessage): + visible_text.extend( + content.text + for content in item.content + if isinstance(content, ResponseOutputText) + ) + elif not isinstance(item, ResponseReasoningItem): + has_tool_calls = True + + terminal = classify_harmony_terminal( + content="\n".join(visible_text), + has_tool_calls=has_tool_calls, + token_ids=token_ids, + ) + if terminal.state not in { + HarmonyTerminalState.RECOVERED_CONTENT, + HarmonyTerminalState.CONTENT_NULL, + }: + return output_items, terminal + + # Never retain completed message items with empty visible content. + filtered_items = [ + item + for item in output_items + if not ( + isinstance(item, ResponseOutputMessage) + and not any( + isinstance(content, ResponseOutputText) and content.text.strip() + for content in item.content + ) + ) + ] + if terminal.state is HarmonyTerminalState.RECOVERED_CONTENT: + assert terminal.content is not None + recovered_message = Message.from_role_and_content( + Role.ASSISTANT, terminal.content + ).with_channel("final") + filtered_items.append(_parse_final_message(recovered_message)) + return filtered_items, terminal diff --git a/vllm/entrypoints/openai/responses/protocol.py b/vllm/entrypoints/openai/responses/protocol.py index 30a920663651..eca1df404f86 100644 --- a/vllm/entrypoints/openai/responses/protocol.py +++ b/vllm/entrypoints/openai/responses/protocol.py @@ -38,6 +38,9 @@ ResponseCompletedEvent as OpenAIResponseCompletedEvent, ) from openai.types.responses import ResponseCreatedEvent as OpenAIResponseCreatedEvent +from openai.types.responses import ( + ResponseIncompleteEvent as OpenAIResponseIncompleteEvent, +) from openai.types.responses import ( ResponseInProgressEvent as OpenAIResponseInProgressEvent, ) @@ -621,6 +624,9 @@ class ResponsesResponse(OpenAIBaseModel): reasoning: Reasoning | None = None service_tier: Literal["auto", "default", "flex", "scale", "priority"] status: ResponseStatus + # vLLM extension used to distinguish parser-incomplete output from a real + # max-output-token truncation while retaining the standard Responses status. + stop_reason: str | int | None = None text: ResponseTextConfig | None = None top_logprobs: int | None = None truncation: Literal["auto", "disabled"] @@ -695,6 +701,7 @@ def from_request( input_messages: ResponseInputOutputMessage | None = None, output_messages: ResponseInputOutputMessage | None = None, kv_transfer_params: dict[str, Any] | None = None, + stop_reason: str | int | None = None, ) -> "ResponsesResponse": incomplete_details: IncompleteDetails | None = None if status == "incomplete": @@ -727,6 +734,7 @@ def from_request( presence_penalty=sampling_params.presence_penalty, frequency_penalty=sampling_params.frequency_penalty, status=status, + stop_reason=stop_reason, text=request.text, top_logprobs=sampling_params.logprobs, truncation=request.truncation, @@ -794,10 +802,15 @@ class ResponseInProgressEvent(OpenAIResponseInProgressEvent): response: ResponsesResponse # type: ignore[override] +class ResponseIncompleteEvent(OpenAIResponseIncompleteEvent): + response: ResponsesResponse # type: ignore[override] + + StreamingResponsesResponse: TypeAlias = ( ResponseCreatedEvent | ResponseInProgressEvent | ResponseCompletedEvent + | ResponseIncompleteEvent | ResponseOutputItemAddedEvent | ResponseOutputItemDoneEvent | ResponseContentPartAddedEvent diff --git a/vllm/entrypoints/openai/responses/serving.py b/vllm/entrypoints/openai/responses/serving.py index 09c902008a48..78c972cc5b92 100644 --- a/vllm/entrypoints/openai/responses/serving.py +++ b/vllm/entrypoints/openai/responses/serving.py @@ -9,7 +9,7 @@ from contextlib import AsyncExitStack from copy import copy from http import HTTPStatus -from typing import Any, Final, cast +from typing import Any, Final from fastapi import Request from openai.types.responses import ( @@ -24,6 +24,7 @@ from openai.types.responses.response_output_text import Logprob, LogprobTopLogprob from openai.types.responses.tool import Mcp, Tool from openai_harmony import Message as OpenAIHarmonyMessage +from pydantic import TypeAdapter from vllm import envs from vllm.config.utils import replace @@ -47,6 +48,9 @@ ) from vllm.entrypoints.openai.models.serving import OpenAIServingModels from vllm.entrypoints.openai.parser.harmony_utils import ( + CONTENT_NULL_STOP_REASON, + HarmonyTerminalResult, + HarmonyTerminalState, build_harmony_preamble, extract_instructions_from_messages, get_encoding, @@ -62,6 +66,7 @@ StreamingHarmonyContext, ) from vllm.entrypoints.openai.responses.harmony import ( + apply_harmony_terminal_invariant, construct_harmony_previous_input_messages, harmony_to_response_output, parser_state_to_response_output, @@ -72,6 +77,7 @@ OutputTokensDetails, ResponseCompletedEvent, ResponseCreatedEvent, + ResponseIncompleteEvent, ResponseInProgressEvent, ResponseInputOutputItem, ResponseInputOutputMessage, @@ -86,6 +92,8 @@ _StateType, emit_content_delta_events, emit_previous_item_done_events, + emit_text_delta_events, + emit_text_output_done_events, emit_tool_action_events, split_delta, ) @@ -954,6 +962,7 @@ async def responses_full_generator( # we guarantee that if the status is not "completed", it is accurate. # "completed" is implemented as the "catch-all" for now. status: ResponseStatus = "completed" + stop_reason_: str | int | None = None input_messages: ResponseInputOutputMessage | None = None output_messages: ResponseInputOutputMessage | None = None @@ -984,11 +993,17 @@ async def responses_full_generator( elif self.use_harmony: assert isinstance(context, HarmonyContext) output = self._make_response_output_items_with_harmony(context) + output, terminal = self._finalize_harmony_output_items( + context, output, request.request_id + ) if request.enable_response_messages: input_messages = context.messages[: context.num_init_messages] output_messages = context.messages[context.num_init_messages :] num_tool_output_tokens = context.num_tool_output_tokens - if len(output) > 0: + if terminal.state is HarmonyTerminalState.CONTENT_NULL: + status = "incomplete" + stop_reason_ = CONTENT_NULL_STOP_REASON + elif len(output) > 0: if context.finish_reason == "length": status = "incomplete" elif context.finish_reason == "abort": @@ -1093,6 +1108,7 @@ async def responses_full_generator( status=status, usage=usage, kv_transfer_params=context.kv_transfer_params, + stop_reason=stop_reason_, ) if request.store: @@ -1270,6 +1286,32 @@ def _make_response_output_items_with_harmony( output_items.extend(last_items) return output_items + def _finalize_harmony_output_items( + self, + context: HarmonyContext, + output_items: list[ResponseOutputItem], + request_id: str, + *, + log_content_null: bool = True, + ) -> tuple[list[ResponseOutputItem], HarmonyTerminalResult]: + output_items, terminal = apply_harmony_terminal_invariant( + output_items, context.output_token_ids + ) + if terminal.state is HarmonyTerminalState.RECOVERED_CONTENT: + assert terminal.content is not None + context.record_terminal_recovery(terminal.content) + logger.warning( + "Recovered filtered Harmony content for Responses request %s.", + request_id, + ) + elif terminal.state is HarmonyTerminalState.CONTENT_NULL and log_content_null: + logger.warning( + "Harmony produced no recoverable visible Responses content for " + "request %s; marking it content_null.", + request_id, + ) + return output_items, terminal + def _get_harmony_builtin_tool_descriptions( self, request: ResponsesRequest, tool_types: set[str] ) -> dict[str, str | None]: @@ -1459,7 +1501,10 @@ async def responses_background_stream_generator( while current_index < len(event_deque): event = event_deque[current_index] yield event - if getattr(event, "type", "unknown") == "response.completed": + if getattr(event, "type", "unknown") in { + "response.completed", + "response.incomplete", + }: return current_index += 1 @@ -1665,22 +1710,38 @@ async def _process_harmony_streaming_events( # finish_reason='error' indicates a retryable error self._raise_if_error(ctx.finish_reason, request.request_id) - if ctx.is_expecting_start(): - if len(ctx.parser.messages) > 0: - previous_item = ctx.parser.messages[-1] - for event in emit_previous_item_done_events( - previous_item, state, ctx.function_tool_names - ): - yield _increment_sequence_number_and_return(event) - state.reset_for_new_item() + if not ctx.harmony_parse_failed: + if ctx.is_expecting_start(): + if len(ctx.parser.messages) > 0: + previous_item = ctx.parser.messages[-1] + for event in emit_previous_item_done_events( + previous_item, state, ctx.function_tool_names + ): + yield _increment_sequence_number_and_return(event) + state.reset_for_new_item() + + # Stream the output of a harmony message + for event in emit_content_delta_events(ctx, state): + yield _increment_sequence_number_and_return(event) - # Stream the output of a harmony message - for event in emit_content_delta_events(ctx, state): - yield _increment_sequence_number_and_return(event) + # Stream tool call outputs + for event in emit_tool_action_events(ctx, state, self.tool_server): + yield _increment_sequence_number_and_return(event) - # Stream tool call outputs - for event in emit_tool_action_events(ctx, state, self.tool_server): - yield _increment_sequence_number_and_return(event) + if ctx.last_output_finished: + output_items = self._make_response_output_items_with_harmony(ctx) + _, terminal = self._finalize_harmony_output_items( + ctx, + output_items, + request.request_id, + log_content_null=False, + ) + if terminal.state is HarmonyTerminalState.RECOVERED_CONTENT: + assert terminal.content is not None + for event in emit_text_delta_events(terminal.content, state): + yield _increment_sequence_number_and_return(event) + for event in emit_text_output_done_events(terminal.content, state): + yield _increment_sequence_number_and_return(event) async def responses_stream_generator( self, @@ -1783,10 +1844,17 @@ async def empty_async_generator(): request_metadata, created_time=created_time, ) - yield _increment_sequence_number_and_return( - ResponseCompletedEvent( + assert isinstance(final_response, ResponsesResponse) + if final_response.status == "incomplete": + terminal_event: StreamingResponsesResponse = ResponseIncompleteEvent( + type="response.incomplete", + sequence_number=-1, + response=final_response, + ) + else: + terminal_event = ResponseCompletedEvent( type="response.completed", sequence_number=-1, response=final_response, ) - ) + yield _increment_sequence_number_and_return(terminal_event) From 314b49459b1ea257af5a15a5dc51216f6ccf6f1c Mon Sep 17 00:00:00 2001 From: ankrovv Date: Mon, 6 Jul 2026 13:18:07 -0400 Subject: [PATCH 3/3] fix(gpt-oss): normalize constrained Harmony recipient A constrained final message (<|channel|>final<|constrain|>json<|message|>) can be parsed by openai-harmony with recipient="<|constrain|>json", which the Responses converter routed to an mcp_call - leaking the control token into name/server_label and putting the JSON answer in arguments instead of output_text. Strip the <|constrain|> marker from the recipient before dispatch so the message is returned as normal output. Backport of upstream #45657 (issue #45570), adapted to the old-layout responses/harmony.py converter. --- .../test_harmony_recipient_normalization.py | 52 +++++++++++++++++++ vllm/entrypoints/openai/responses/harmony.py | 27 +++++++++- 2 files changed, 78 insertions(+), 1 deletion(-) create mode 100644 tests/entrypoints/openai/responses/test_harmony_recipient_normalization.py diff --git a/tests/entrypoints/openai/responses/test_harmony_recipient_normalization.py b/tests/entrypoints/openai/responses/test_harmony_recipient_normalization.py new file mode 100644 index 000000000000..d53096a615fb --- /dev/null +++ b/tests/entrypoints/openai/responses/test_harmony_recipient_normalization.py @@ -0,0 +1,52 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM project +"""Unit tests for constrained-recipient normalization in the Harmony -> Responses +converter. + +Regression coverage for vllm-project/vllm#45570: a constrained ``final`` message +(``<|channel|>final<|constrain|>json<|message|>{...}``) can be parsed by +openai-harmony with ``recipient == "<|constrain|>json"``. Without normalization +the converter routes it to :func:`_parse_mcp_call` and leaks the control token +into an ``mcp_call`` item's ``name``/``server_label``. It must instead be +returned as a normal output message. +""" + +from openai_harmony import Message, Role + +from vllm.entrypoints.openai.responses.harmony import ( + _normalize_recipient, + harmony_to_response_output, +) + + +def test_normalize_recipient_strips_constrain_marker(): + # Bare marker -> no real recipient. + assert _normalize_recipient("<|constrain|>json") is None + # Marker after a real recipient -> keep the real part. + assert ( + _normalize_recipient("functions.get_weather <|constrain|>json") + == "functions.get_weather" + ) + # Untainted recipients pass through unchanged. + assert _normalize_recipient("functions.get_weather") == "functions.get_weather" + assert _normalize_recipient("repo_browser.list") == "repo_browser.list" + assert _normalize_recipient(None) is None + assert _normalize_recipient("") == "" + + +def test_constrained_final_message_not_parsed_as_mcp_call(): + """A final message whose only 'recipient' is the leaked <|constrain|> marker + must become a message, not an mcp_call (vllm-project/vllm#45570).""" + payload = '{"name": "Science Fair", "date": "Friday"}' + msg = ( + Message.from_role_and_content(Role.ASSISTANT, payload) + .with_channel("final") + .with_recipient("<|constrain|>json") + ) + + items = harmony_to_response_output(msg) + + assert len(items) == 1 + assert items[0].type == "message" + assert items[0].content[0].text == payload + assert all(getattr(item, "type", None) != "mcp_call" for item in items) diff --git a/vllm/entrypoints/openai/responses/harmony.py b/vllm/entrypoints/openai/responses/harmony.py index c96733cd1874..5dea6918fce2 100644 --- a/vllm/entrypoints/openai/responses/harmony.py +++ b/vllm/entrypoints/openai/responses/harmony.py @@ -364,6 +364,31 @@ def _parse_final_message(message: Message) -> ResponseOutputItem: ) +def _normalize_recipient(recipient: str | None) -> str | None: + """Strip a leaked ``<|constrain|>`` content-type marker from a recipient. + + openai-harmony can mis-read the ``<|constrain|>json`` content-type of a + constrained ``final`` message as the message recipient (upstream issue + vllm-project/vllm#45570; fixed upstream by #45657). Left untouched, such a + recipient is routed to :func:`_parse_mcp_call` and the control token leaks + into an ``mcp_call`` item's ``name``/``server_label``. Strip the marker; if + nothing real precedes it, there is no recipient. + + Args: + recipient: The raw recipient string from the Harmony message. + + Returns: + The recipient with any trailing ``<|constrain|>...`` removed, or + ``None`` when only the marker remained. + """ + if not recipient: + return recipient + constrain_index = recipient.find("<|constrain|>") + if constrain_index == -1: + return recipient + return recipient[:constrain_index].rstrip() or None + + def _parse_mcp_recipient(recipient: str) -> tuple[str, str]: """Parse MCP recipient into (server_label, tool_name). @@ -443,7 +468,7 @@ def harmony_to_response_output( return [] output_items: list[ResponseOutputItem] = [] - recipient = message.recipient + recipient = _normalize_recipient(message.recipient) if recipient is not None: # Browser tool calls (browser.search, browser.open, browser.find)