[None][fix] Responses: final snapshot keeps the status the stream reported

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)
 

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.", "</think>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.</think>" + _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"