[TRTLLM-17273][chore] Read result completion through the public finished property in serve and scaffolding

tensorrt_llm/executor/result.py

@@ -334,6 +334,11 @@ def error(self) -> Optional[str]:
         """Return the error message if this result completed with an error."""
         return self._error_msg
 
+    @property
+    def finished(self) -> bool:
+        """Whether the whole request has finished. A read-only view of ``_done``."""
+        return self._done
+
     def _maybe_fill_spec_dec_perf_metrics(
             self, perf_metrics: "tllm.RequestPerfMetrics") -> None:
         """Backfill RequestPerfMetrics.speculative_decoding in the PyTorch flow.
@@ -1145,10 +1150,6 @@ def abort(self) -> None:
         self._executor().abort_request(self.request_id)
         super().abort()
 
-    @property
-    def finished(self) -> bool:
-        return self._done
-
     def clear_logprob_params(self) -> None:
         # Remove temporary attribute used in executor
         # for a cleaner external-facing output.

tensorrt_llm/scaffolding/contrib/AsyncGeneration/__init__.py

@@ -0,0 +1,2 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0

tensorrt_llm/scaffolding/contrib/AsyncGeneration/stream_generation.py

@@ -43,12 +43,11 @@ async def get_step_or_more_tokens(task: StreamGenerationTask):
             return TaskStatus.SUCCESS
 
         for _ in range(task.streaming_step):
-            await task.request_handle._aresult_step()
-            if task.request_handle._done:
+            await anext(task.request_handle)
+            if task.request_handle.finished:
                 break
-        while not task.request_handle._done:
-            async_task = asyncio.create_task(
-                task.request_handle._aresult_step())
+        while not task.request_handle.finished:
+            async_task = asyncio.create_task(anext(task.request_handle))
             if not async_task.done():
                 async_task.cancel()
                 break
@@ -58,7 +57,7 @@ async def get_step_or_more_tokens(task: StreamGenerationTask):
         task.cumulative_logprob = task.request_handle.outputs[
             0].cumulative_logprob
         task.logprobs = task.request_handle.outputs[0].logprobs
-        if task.request_handle._done:
+        if task.request_handle.finished:
             task.end_flag = True
 
     sampling_params = worker.convert_task_params(task)

tensorrt_llm/scaffolding/worker.py

@@ -598,16 +598,16 @@ def convert_task_params(self, task: GenerationTask):
     async def streaming_generate_helper(self, generate_result, step_at_least,
                                         streaming_output_list):
         step = 0
-        while not generate_result._done:
-            async_task = asyncio.create_task(generate_result._aresult_step())
+        while not generate_result.finished:
+            async_task = asyncio.create_task(anext(generate_result))
             if step_at_least and step >= step_at_least and not async_task.done(
             ):
                 async_task.cancel()
                 break
             await async_task
             step += 1
             # do not put the last token to the streaming_output_list
-            if streaming_output_list is not None and not generate_result._done:
+            if streaming_output_list is not None and not generate_result.finished:
                 streaming_output_list.append(
                     ScaffoldingOutput(
                         generate_result.outputs[0].text,
@@ -656,7 +656,7 @@ async def stream_generation_handler(
 
         self.fill_task_with_result(task, task.request_handle)
 
-        if task.request_handle._done:
+        if task.request_handle.finished:
             task.end_flag = True
         return TaskStatus.SUCCESS
 

tensorrt_llm/serve/postprocess_handlers.py

@@ -624,7 +624,7 @@ def yield_first_chat(num_tokens: int,
         data = chunk.model_dump_json(exclude_none=True)
         res.append(f"data: {data}\n\n")
 
-    if include_usage and rsp._done:
+    if include_usage and rsp.finished:
         completion_tokens = sum(output.length for output in rsp.outputs)
         final_usage = UsageInfo(
             prompt_tokens=prompt_tokens,
@@ -915,7 +915,7 @@ def completion_stream_post_processor(rsp: DetokenizedGenerationResultBase,
         data = chunk.model_dump_json(exclude_unset=False)
         res.append(f"data: {data}\n\n")
 
-    if include_usage and rsp._done:
+    if include_usage and rsp.finished:
         completion_tokens = sum(output.length for output in rsp.outputs)
         final_usage = UsageInfo(
             prompt_tokens=prompt_tokens,
@@ -1054,7 +1054,7 @@ def chat_harmony_streaming_post_processor(
         result=rsp,
         model=args.model,
         request_id=str(rsp.id),
-        done=rsp._done,
+        done=rsp.finished,
         num_prompt_tokens=prompt_tokens,
         first_iteration=args.first_iteration,
         stream_options=args.stream_options,
@@ -1101,7 +1101,7 @@ def responses_api_streaming_post_processor(
         raise ValueError(
             "streaming_processor is required for streaming post-processing")
     outputs = args.streaming_processor.process_single_output(rsp)
-    if rsp._done:
+    if rsp.finished:
         outputs.append(
             args.streaming_processor.get_final_response_non_store(
                 rsp, args.num_prompt_tokens, tokenizer=args.tokenizer))

tensorrt_llm/serve/responses_utils.py

@@ -3092,7 +3092,7 @@ def process_single_output(self, res: GenerationResult) -> list[str]:
             event_generator = _generate_streaming_event(
                 output=output,
                 request=self.request,
-                finished_generation=res._done,
+                finished_generation=res.finished,
                 streaming_events_helper=self.streaming_events_helper,
                 reasoning_parser_id=self.reasoning_parser,
                 tool_parser_id=self.tool_parser,

tests/integration/test_lists/test-db/l0_cpu.yml

@@ -295,6 +295,7 @@ l0_cpu:
   - unittest/quantization/test_mode.py
   - unittest/scaffolding
   - unittest/scripts
+  - unittest/serve/test_postprocess_finished.py
   - unittest/tools
   - unittest/usage
   - unittest/utils/test_logger.py

tests/unittest/llmapi/apps/test_harmony_parsing.py

@@ -117,7 +117,7 @@ def _make_mock_result(
 
     result = Mock()
     result.outputs = [output]
-    result._done = True
+    result.finished = True
     result.cached_tokens = cached_tokens
     return result
 
@@ -650,7 +650,7 @@ class TestStreamMetadata:
     def test_postprocessor_passes_cached_stream_metadata(self):
         mock_result = _make_mock_result(token_ids_diff=[1, 2, 3])
         mock_result.id = 456
-        mock_result._done = False
+        mock_result.finished = False
         args = ChatCompletionPostprocArgs(
             model="test-model",
             tools=[],

tests/unittest/llmapi/apps/test_responses_stream_final_consistency.py

@@ -81,7 +81,9 @@ def _generation_chunk(accumulated, diff, done, finish_reason="stop"):
         token_ids=[1, 2, 3],
         disaggregated_params=None,
     )
-    return SimpleNamespace(outputs=[output], _done=done, prompt_token_ids=[1, 2], cached_tokens=0)
+    return SimpleNamespace(
+        outputs=[output], finished=done, prompt_token_ids=[1, 2], cached_tokens=0
+    )
 
 
 def _stream(processor, chunks):

tests/unittest/llmapi/apps/test_responses_streaming_events.py

@@ -523,7 +523,7 @@ def test_streamed_responses_use_wire_field_names():
         token_ids=[1],
         disaggregated_params=None,
     )
-    result = SimpleNamespace(outputs=[output], _done=True, prompt_token_ids=[1], cached_tokens=0)
+    result = SimpleNamespace(outputs=[output], finished=True, prompt_token_ids=[1], cached_tokens=0)
     frames = processor.get_initial_responses() + processor.process_single_output(result)
     frames.append(processor.get_final_response_non_store(result))
 
@@ -548,9 +548,7 @@ def _step(text, diff, done):
         token_ids=[1] * len(text),
         disaggregated_params=None,
     )
-    return SimpleNamespace(
-        outputs=[output], _done=done, prompt_token_ids=[1], cached_tokens=0, finished=done
-    )
+    return SimpleNamespace(outputs=[output], finished=done, prompt_token_ids=[1], cached_tokens=0)
 
 
 def _server_frames(steps, processor=None, fail=None, produced=None):
@@ -573,7 +571,7 @@ async def frames():
             text += diff
             result = _step(text, diff, done=fail is None and i == len(steps) - 1)
             step_frames = processor.process_single_output(result)
-            if result._done:
+            if result.finished:
                 step_frames.append(processor.get_final_response_non_store(result))
             for frame in step_frames:
                 produced.append(frame)
@@ -972,7 +970,7 @@ async def __aiter__(self):
                 await asyncio.sleep(pause)
                 text += diff
                 result = _step(text, diff, done=i == len(steps) - 1)
-                self.finished = result._done
+                self.finished = result.finished
                 yield result
 
     server.generator = SimpleNamespace(

tests/unittest/llmapi/apps/test_responses_usage.py

@@ -241,7 +241,7 @@ def test_a_streamed_response_counts_reasoning_tokens():
         token_ids=[1, 2, 3],
         disaggregated_params=None,
     )
-    result = SimpleNamespace(outputs=[output], _done=True, prompt_token_ids=[1], cached_tokens=0)
+    result = SimpleNamespace(outputs=[output], finished=True, prompt_token_ids=[1], cached_tokens=0)
     processor.process_single_output(result)
     frame = processor.get_final_response_non_store(result, tokenizer=_WordTokenizer())
     usage = json.loads(frame.split("data: ", 1)[1])["response"]["usage"]

tests/unittest/llmapi/apps/test_tool_parsers.py

@@ -5841,7 +5841,7 @@ def _rsp(text, finish_reason):
 
         rsp = Mock()
         rsp.outputs = [output]
-        rsp._done = finish_reason is not None
+        rsp.finished = finish_reason is not None
         rsp.cached_tokens = 0
         rsp.prompt_token_ids = [1, 2, 3]
         # Read via getattr() and fed straight into a pydantic model, so it must
@@ -6370,7 +6370,7 @@ def _stream_deltas(args, chunks, finish_reason="stop"):
             rsp = SimpleNamespace(outputs=[output],
                                   cached_tokens=0,
                                   id="test-request",
-                                  _done=last)
+                                  finished=last)
             for line in chat_stream_post_processor(rsp, args):
                 payloads.append(json.loads(line[len("data: "):]))
         return payloads

tests/unittest/llmapi/test_executor.py

@@ -109,10 +109,16 @@ def test_GenerationResultBase():
     )
     result._handle_response(create_rsp(2, finished=False))
     result._handle_response(create_rsp(3, finished=False))
+    assert not result.finished
     result._handle_response(create_rsp(4, finished=True))
     print(result.outputs[0])
     assert len(result.outputs[0].token_ids) == 3
     assert result._done
+    assert result.finished
+    # finished is a read-only view of the completion flag; only `_done` is
+    # ever written (by the response-handling paths), never the property.
+    with pytest.raises(AttributeError):
+        result.finished = False
 
 
 @pytest.mark.parametrize("pytorch", [False, True])

tests/unittest/scaffolding/test_worker_streaming_step.py

@@ -0,0 +1,136 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+"""Scaffolding streaming helpers use only the public result surface.
+
+That surface is the ``finished`` property and async iteration (``anext``).
+The fake handle below deliberately implements nothing else, so a regression
+back to private members (``_done``, ``_aresult_step``) fails here.
+"""
+
+import asyncio
+
+import pytest
+
+from tensorrt_llm.scaffolding import TaskStatus
+from tensorrt_llm.scaffolding.contrib.AsyncGeneration.stream_generation import (
+    StreamGenerationTask as ContribStreamGenerationTask,
+)
+from tensorrt_llm.scaffolding.contrib.AsyncGeneration.stream_generation import (
+    stream_generation_handler,
+)
+from tensorrt_llm.scaffolding.task import StreamGenerationTask
+from tensorrt_llm.scaffolding.worker import TRTLLMWorker
+
+pytestmark = pytest.mark.cpu_only
+
+
+class _FakeOutput:
+    def __init__(self):
+        self.text = ""
+        self.token_ids = []
+        self.cumulative_logprob = None
+        self.logprobs = []
+
+
+class _FakeStreamingHandle:
+    """A streaming generation handle exposing only the public surface.
+
+    One generated token per async-iteration step; ``finished`` once
+    ``total_steps`` tokens are out. No ``_done``, no ``_aresult_step``.
+    """
+
+    def __init__(self, total_steps):
+        self.total_steps = total_steps
+        self.steps_taken = 0
+        self.outputs = [_FakeOutput()]
+        self.context_logits = None
+
+    @property
+    def finished(self) -> bool:
+        return self.steps_taken >= self.total_steps
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if self.finished:
+            raise StopAsyncIteration
+        await asyncio.sleep(0)
+        self.steps_taken += 1
+        self.outputs[0].text += "t"
+        self.outputs[0].token_ids = self.outputs[0].token_ids + [self.steps_taken]
+        return self
+
+
+def test_streaming_generate_helper_steps_until_finished():
+    worker = TRTLLMWorker(llm=None, tokenizer=None)
+    handle = _FakeStreamingHandle(total_steps=3)
+    collected = []
+
+    asyncio.run(worker.streaming_generate_helper(handle, None, collected))
+
+    assert handle.finished
+    assert handle.steps_taken == 3
+    # One ScaffoldingOutput per non-final step; the last token is not queued.
+    assert [out.text for out in collected] == ["t", "tt"]
+    assert [out.token_ids for out in collected] == [[1], [1, 2]]
+
+
+def test_streaming_generate_helper_stops_early_at_step_at_least():
+    worker = TRTLLMWorker(llm=None, tokenizer=None)
+    handle = _FakeStreamingHandle(total_steps=10)
+    collected = []
+
+    asyncio.run(worker.streaming_generate_helper(handle, 2, collected))
+
+    assert not handle.finished
+    assert handle.steps_taken == 2
+    assert [out.text for out in collected] == ["t", "tt"]
+
+
+def test_worker_stream_generation_handler_sets_end_flag_when_finished():
+    worker = TRTLLMWorker(llm=None, tokenizer=None)
+    task = StreamGenerationTask()
+    task.streaming_step = 5
+    task.request_handle = _FakeStreamingHandle(total_steps=2)
+
+    status = asyncio.run(worker.stream_generation_handler(task))
+
+    assert status == TaskStatus.SUCCESS
+    assert task.end_flag
+    assert task.output_str == "tt"
+    assert task.output_tokens == [1, 2]
+
+
+class _FakeContribWorker:
+    """Only what the AsyncGeneration handler calls on a worker."""
+
+    def convert_task_params(self, task):
+        return None
+
+
+def test_contrib_stream_generation_handler_sets_end_flag_when_finished():
+    task = ContribStreamGenerationTask()
+    task.streaming_step = 5
+    task.request_handle = _FakeStreamingHandle(total_steps=3)
+
+    status = asyncio.run(stream_generation_handler(_FakeContribWorker(), task))
+
+    assert status == TaskStatus.SUCCESS
+    assert task.end_flag
+    assert task.output_str == "ttt"
+    assert task.output_tokens == [1, 2, 3]
+
+
+def test_contrib_stream_generation_handler_waits_for_more_steps():
+    task = ContribStreamGenerationTask()
+    task.streaming_step = 2
+    task.request_handle = _FakeStreamingHandle(total_steps=5)
+
+    status = asyncio.run(stream_generation_handler(_FakeContribWorker(), task))
+
+    assert status == TaskStatus.SUCCESS
+    assert not task.end_flag
+    assert task.request_handle.steps_taken == 2
+    assert task.output_str == "tt"
+    assert task.output_tokens == [1, 2]