diff --git a/tensorrt_llm/serve/responses_utils.py b/tensorrt_llm/serve/responses_utils.py index f7d93a1471de..0f9718b36e0d 100644 --- a/tensorrt_llm/serve/responses_utils.py +++ b/tensorrt_llm/serve/responses_utils.py @@ -662,7 +662,7 @@ def _parse_output_message_harmony(message: Message) -> list[ResponseOutputItem]: summary=[], type="reasoning", content=[Content(text=content.text, type="reasoning_text")], - status=None, + status="completed", ) output_items.append(reasoning_item) elif message.channel == "commentary": @@ -677,6 +677,7 @@ def _parse_output_message_harmony(message: Message) -> list[ResponseOutputItem]: type="function_call", name=function_name, id=f"fc_{_random_uuid()}", + status="completed", ) output_items.append(response_item) elif message.recipient.startswith( @@ -687,7 +688,7 @@ def _parse_output_message_harmony(message: Message) -> list[ResponseOutputItem]: summary=[], type="reasoning", content=[Content(text=content.text, type="reasoning_text")], - status=None, + status="completed", ) output_items.append(reasoning_item) else: @@ -1895,7 +1896,9 @@ def _streamed_or_fresh_id(item_type: str) -> str: content=[ Content(text=record.text, type="reasoning_text") ], - status=None, + # The stream closed this item as completed; the + # snapshot reports the same status. + status="completed", )) else: output_items.append( @@ -1928,7 +1931,7 @@ def _streamed_or_fresh_id(item_type: str) -> str: content=[ Content(text=reasoning_text, type="reasoning_text") ], - status=None, + status="completed", )) stored_reasoning = reasoning_text @@ -1955,8 +1958,12 @@ def _streamed_or_fresh_id(item_type: str) -> str: tool_calls_item = list(streamed_tool_calls) else: tool_resolution = _tool_resolution(tools) + # As streamed: the stream delivers every call it announces as + # completed. tool_calls_item = [ - _tool_call_output_item(call, tool_resolution) for call in calls + _tool_call_output_item(call, + tool_resolution, + status="completed") for call in calls ] output_items.extend(tool_calls_item) diff --git a/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py b/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py index e7e2ceca84cb..bf789648c8e2 100644 --- a/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py +++ b/tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py @@ -32,6 +32,7 @@ ResponsesStreamingProcessor, StreamedItem, _create_output_content, + _parse_output_message_harmony, _whole_text_tool_calls, create_response_non_store, ) @@ -385,3 +386,67 @@ def test_the_stream_releases_what_the_whole_text_parse_reads_beyond_it( SimpleNamespace(index=0, text=whole), request, helper, None, "qwen3", [] ) assert (released, calls, logger.warning.called) == (owed, [], warned) + + +# --------------------------------------------------------------------------- +# Item status: the snapshot reports what the stream closed each item with +# --------------------------------------------------------------------------- + + +def test_final_items_report_the_status_the_stream_closed_them_with(): + """Reasoning and function-call items stay `completed` in the snapshot. + + Their done events say `completed`; a snapshot that built them without a + status read `null` for the same items, so a consumer that filters items by + status reconstructed a different turn from each view. + """ + processor = _processor(tools=[_tool()], reasoning_parser="glm47", tool_parser="glm47") + frames, result = _stream(processor, ["Plan it.", "Calling: ", _EXEC_CALL]) + + streamed = {item["type"]: item.get("status") for item in _done_items(frames)} + final = {item["type"]: item.get("status") for item in _final_output(processor, result)} + + assert streamed == { + "reasoning": "completed", + "message": "completed", + "function_call": "completed", + } + assert final == streamed + + +def test_non_streaming_items_are_completed_too(): + """No stream ran; the body is the only account, and it says completed.""" + items, _messages, _reasoning = _create_output_content( + SimpleNamespace(outputs=[SimpleNamespace(index=0, text="Plan it." + _EXEC_CALL)]), + reasoning_parser="glm47", + tool_parser="glm47", + tools=[_tool()], + streamed_item_ids=None, + ) + assert [(item.type, item.status) for item in items] == [ + ("reasoning", "completed"), + ("function_call", "completed"), + ] + + +@pytest.mark.parametrize( + "channel, recipient, item_type", + [ + ("analysis", None, "reasoning"), + ("commentary", "functions.get_weather", "function_call"), + ("commentary", "python", "reasoning"), + ], + ids=["analysis", "function_call", "python_commentary"], +) +def test_harmony_reasoning_and_function_call_items_are_completed(channel, recipient, item_type): + openai_harmony = pytest.importorskip("openai_harmony") + message = openai_harmony.Message.from_role_and_content( + openai_harmony.Role.ASSISTANT, '{"city": "Rome"}' + ).with_channel(channel) + if recipient is not None: + message = message.with_recipient(recipient) + + (item,) = _parse_output_message_harmony(message) + + assert item.type == item_type + assert item.status == "completed"