test(trtllm): enable fault tolerance coverage

components/src/dynamo/trtllm/request_handlers/handler_base.py

@@ -208,7 +208,7 @@ def abort(self) -> None:
         """Abort immediately if first token received, otherwise defer."""
         if self._first_token_received:
             self._generation_result.abort()
-            logging.debug("Deferred abort: first token already received, aborting now")
+            logging.debug("Deferred abort: immediate path, engine abort fired")
         else:
             logging.debug(
                 "Deferred abort: first token not received, spawning background task"
@@ -223,7 +223,7 @@ async def _wait_and_abort(self) -> None:
         except Exception:
             pass
         self._generation_result.abort()
-        logging.debug("Deferred abort: background task completed, abort fired")
+        logging.debug("Deferred abort: deferred path, engine abort fired")
 
 
 @dataclass
@@ -906,12 +906,15 @@ def _request_has_multimodal(self, request: dict) -> bool:
                         return True
         return False
 
-    def _normalize_request_format(self, request: dict) -> None:
+    @staticmethod
+    def _normalize_request_format(request: dict) -> None:
         """
         Convert OpenAI request format to TRT-LLM internal format.
 
         Moves fields from OpenAI locations to where TRT-LLM expects them:
         - max_tokens: top-level → stop_conditions.max_tokens
+        - min_tokens: top-level → stop_conditions.min_tokens
+        - ignore_eos: top-level → stop_conditions.ignore_eos
         - temperature: top-level → sampling_options.temperature
 
         Note: The Rust frontend's PrefillRouter handles the *value* of max_tokens
@@ -924,17 +927,17 @@ def _normalize_request_format(self, request: dict) -> None:
         # Ensure stop_conditions exists
         if "stop_conditions" not in request:
             request["stop_conditions"] = {}
-        if "max_tokens" in request and "max_tokens" not in request["stop_conditions"]:
-            request["stop_conditions"]["max_tokens"] = request.pop("max_tokens")
+        for field in ("max_tokens", "min_tokens", "ignore_eos"):
+            if field in request:
+                value = request.pop(field)
+                request["stop_conditions"].setdefault(field, value)
 
         # Ensure sampling_options exists
         if "sampling_options" not in request:
             request["sampling_options"] = {}
-        if (
-            "temperature" in request
-            and "temperature" not in request["sampling_options"]
-        ):
-            request["sampling_options"]["temperature"] = request.pop("temperature")
+        if "temperature" in request:
+            temperature = request.pop("temperature")
+            request["sampling_options"].setdefault("temperature", temperature)
 
     async def _initiate_shutdown(self, error: Exception):
         """Initiate graceful shutdown after fatal error"""

components/src/dynamo/trtllm/tests/test_trtllm_handler_base.py

@@ -211,6 +211,31 @@ def test_post_init_called_when_overriding(self):
         mock_post_init.assert_called_once()
 
 
+class TestNormalizeRequestFormat:
+    def test_moves_openai_stop_fields_without_overwriting_internal_values(self):
+        request = {
+            "max_tokens": 64,
+            "min_tokens": 32,
+            "ignore_eos": True,
+            "temperature": 0.5,
+            "stop_conditions": {"min_tokens": 16},
+            "sampling_options": {"temperature": 0.25},
+        }
+
+        HandlerBase._normalize_request_format(request)
+
+        assert request["stop_conditions"] == {
+            "max_tokens": 64,
+            "min_tokens": 16,
+            "ignore_eos": True,
+        }
+        assert "max_tokens" not in request
+        assert "min_tokens" not in request
+        assert "ignore_eos" not in request
+        assert request["sampling_options"] == {"temperature": 0.25}
+        assert "temperature" not in request
+
+
 class TestGuidedDecodingFromToolChoice:
     """Tests that guided_decoding dicts from Rust are converted to GuidedDecodingParams.
 

tests/fault_tolerance/cancellation/test_trtllm.py

@@ -6,13 +6,12 @@
 - test_request_cancellation_trtllm_aggregated: ~45s (gpu_1)
 - test_request_cancellation_trtllm_decode_cancel: ~65s (gpu_1)
 - test_request_cancellation_trtllm_prefill_cancel: ~65s (gpu_1)
-- test_request_cancellation_trtllm_kv_transfer_cancel: ~65s (gpu_1)
+- test_request_cancellation_trtllm_decode_handoff_cancel: ~65s (gpu_1)
 - Total: ~240s x2 request planes = ~480s (0:08:00)
 """
 
 import logging
 import os
-import shutil
 import time
 
 import pytest
@@ -65,6 +64,7 @@ def __init__(
         request.addfinalizer(lambda port=system_port: deallocate_port(port))
         self.system_port = system_port
         self.frontend_port = frontend_port
+        tmp_path = request.getfixturevalue("tmp_path")
 
         command = [
             "python3",
@@ -80,16 +80,14 @@ def __init__(
             "16384",
         ]
         if mode != "agg":
-            with open("test_request_cancellation_trtllm_config.yaml", "w") as f:
+            config_file = tmp_path / f"trtllm_cancel_config_{system_port}.yaml"
+            with config_file.open("w") as f:
                 f.write(
                     "cache_transceiver_config:\n  backend: DEFAULT\n  max_tokens_in_buffer: 16384\n"
                 )
                 f.write("disable_overlap_scheduler: true\n")
                 f.write("kv_cache_config:\n  max_tokens: 16384\n")
-            command += [
-                "--extra-engine-args",
-                "test_request_cancellation_trtllm_config.yaml",
-            ]
+            command += ["--extra-engine-args", str(config_file)]
 
         health_check_urls = [
             (f"http://localhost:{frontend_port}/v1/models", check_models_api),
@@ -115,16 +113,7 @@ def __init__(
         env["DYN_SYSTEM_USE_ENDPOINT_HEALTH_STATUS"] = '["generate"]'
         env["DYN_SYSTEM_PORT"] = str(system_port)
 
-        # Set log directory based on worker type
-        log_dir = f"{request.node.name}_{mode}_worker"
-
-        # Clean up any existing log directory from previous runs
-        try:
-            shutil.rmtree(log_dir)
-            logger.info(f"Cleaned up existing log directory: {log_dir}")
-        except FileNotFoundError:
-            # Directory doesn't exist, which is fine
-            pass
+        log_dir = tmp_path / f"{mode}_worker"
 
         super().__init__(
             command=command,
@@ -133,7 +122,7 @@ def __init__(
             timeout=300,
             display_output=True,
             terminate_all_matching_process_names=False,
-            log_dir=log_dir,
+            log_dir=str(log_dir),
         )
 
         self.mode = mode
@@ -346,7 +335,9 @@ def test_request_cancellation_trtllm_decode_cancel(
                 )
 
 
-@pytest.mark.skip(reason="TRT-LLM prefill cancellation is disabled due to reliability")
+@pytest.mark.skip(
+    reason="Cancellation does not reach TRT-LLM before prefill completes (1.3.0rc25)"
+)
 @pytest.mark.timeout(195)  # 3x average
 def test_request_cancellation_trtllm_prefill_cancel(
     request, runtime_services_dynamic_ports, predownload_models
@@ -456,20 +447,20 @@ def test_request_cancellation_trtllm_prefill_cancel(
                 )
 
 
-@pytest.mark.skip(reason="Test fails only on CI")
-@pytest.mark.timeout(195)  # 3x average
-def test_request_cancellation_trtllm_kv_transfer_cancel(
+@pytest.mark.timeout(350)  # 3x average
+def test_request_cancellation_trtllm_decode_handoff_cancel(
     request, runtime_services_dynamic_ports, predownload_models
 ):
     """
-    End-to-end test for request cancellation during prefill to decode KV transfer phase.
+    End-to-end test for request cancellation after the decode worker accepts it.
 
-    This test verifies that when a request is cancelled by the client during the KV transfer phase,
-    the system properly handles the cancellation and cleans up resources on the workers.
+    The decode-entry log is emitted before KV transfer begins, so this test does not prove that
+    TRT-LLM supports an engine abort during transfer. It verifies Dynamo's deferred-abort behavior:
+    the engine abort fires after the handoff and both workers remain usable.
 
     Timing (Last Run: 2025-12-09): ~115s total (2 workers at 45% GPU each)
     - Engine initialization: ~92s (frontend: 2s, prefill worker: 45s, decode worker: 45s sequential)
-    - Testing KV transfer cancellation: ~20s
+    - Testing decode handoff cancellation: ~20s
     - Teardown: ~3s
     """
 
@@ -492,9 +483,8 @@ def test_request_cancellation_trtllm_kv_transfer_cancel(
                 # TODO: Why wait after worker ready fixes frontend 404 / 500 flakiness?
                 time.sleep(2)
 
-                # Step 4: Test request cancellation during KV transfer phase
                 logger.info(
-                    "Testing completion request cancellation during KV transfer phase..."
+                    "Testing completion request cancellation after decode handoff..."
                 )
 
                 # Send request with long prompt
@@ -509,24 +499,36 @@ def test_request_cancellation_trtllm_kv_transfer_cancel(
                     match_type="contains",
                 )
 
-                # Poll for decode worker entry signaling start of KV transfer phase
+                # Wait for decode admission. This log precedes KV transfer, and normal
+                # handoff can take longer than poll_for_pattern's 500 ms default.
                 _, decode_log_offset = poll_for_pattern(
                     process=decode_worker,
                     pattern=f"Decode Request ID: {request_id}",
+                    max_wait_ms=30_000,
                     poll_interval_ms=2,
+                    cancellable_request=cancellable_req,
                 )
 
-                # Cancel during KV transfer phase in decode worker
+                # Dynamo defers the engine abort until the first result completes.
                 cancellable_req.cancel()
-                logger.info(
-                    f"Cancelled request ID: {request_id} at beginning of decode"
-                )
+                logger.info(f"Cancelled request ID: {request_id} after decode handoff")
 
                 # Poll for "Aborted Request ID" in decode worker
-                _, decode_log_offset = poll_for_pattern(
+                poll_for_pattern(
                     process=decode_worker,
                     pattern=f"Aborted Request ID: {request_id}",
                     log_offset=decode_log_offset,
+                    max_wait_ms=10_000,
+                )
+
+                # The request-level abort log above is emitted when the wrapper is
+                # invoked. Require the later engine-abort signal so a stuck deferred
+                # task cannot masquerade as successful cancellation.
+                _, decode_log_offset = poll_for_pattern(
+                    process=decode_worker,
+                    pattern="Deferred abort: deferred path, engine abort fired",
+                    log_offset=decode_log_offset,
+                    max_wait_ms=30_000,
                 )
 
                 # Verify frontend log has kill message
@@ -536,23 +538,34 @@ def test_request_cancellation_trtllm_kv_transfer_cancel(
                 )
 
                 logger.info(
-                    "Completion request cancellation at beginning of decode detected successfully"
+                    "Completion request cancellation after decode handoff detected successfully"
                 )
 
                 # Verify the workers are still functional
                 cancellable_req = send_cancellable_request(
-                    frontend.frontend_port, "chat_completion_stream"
+                    frontend.frontend_port,
+                    "chat_completion_stream",
+                    max_tokens=8,
+                    timeout_s=30,
                 )
                 _, decode_log_offset = poll_for_pattern(
                     process=decode_worker,
                     pattern="Decode Request ID: ",
                     log_offset=decode_log_offset,
+                    max_wait_ms=30_000,
                     match_type="contains",
+                    cancellable_request=cancellable_req,
+                )
+                read_streaming_responses(
+                    cancellable_req,
+                    expected_count=1,
+                    deadline_s=30,
+                    drain=True,
+                    require_content=True,
                 )
-                read_streaming_responses(cancellable_req, expected_count=5)
 
                 logger.info(
-                    "Workers are functional after cancellation during KV transfer"
+                    "Workers are functional after cancellation following decode handoff"
                 )
 
                 # Verify cancellation metrics

tests/fault_tolerance/cancellation/test_utils.py

@@ -0,0 +1,71 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Focused tests for cancellation response validation."""
+
+from collections.abc import Iterator
+
+import pytest
+
+from tests.fault_tolerance.cancellation.utils import (
+    CancellableRequest,
+    read_streaming_responses,
+)
+
+pytestmark = [
+    pytest.mark.parallel,
+    pytest.mark.gpu_0,
+    pytest.mark.unit,
+    pytest.mark.pre_merge,
+]
+
+
+class _StreamingResponse:
+    """Minimal response surface consumed by read_streaming_responses()."""
+
+    status_code = 200
+
+    def __init__(self, *lines: bytes):
+        self._lines = lines
+
+    def iter_lines(self) -> Iterator[bytes]:
+        yield from self._lines
+
+
+def _request_with_stream(*lines: bytes) -> CancellableRequest:
+    request = CancellableRequest()
+    request.response = _StreamingResponse(*lines)
+    return request
+
+
+def test_drained_stream_can_require_generated_content():
+    """Role and terminal metadata alone must not prove worker health."""
+    request = _request_with_stream(
+        b'data: {"choices":[{"delta":{"role":"assistant"},"finish_reason":null}]}',
+        b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}',
+        b"data: [DONE]",
+    )
+
+    with pytest.raises(pytest.fail.Exception, match="without generated content"):
+        read_streaming_responses(
+            request,
+            expected_count=1,
+            drain=True,
+            require_content=True,
+        )
+
+
+def test_drained_stream_accepts_generated_content():
+    """A generated text fragment plus terminal metadata proves worker health."""
+    request = _request_with_stream(
+        b'data: {"choices":[{"delta":{"content":"healthy"},"finish_reason":null}]}',
+        b'data: {"choices":[{"delta":{},"finish_reason":"stop"}]}',
+        b"data: [DONE]",
+    )
+
+    read_streaming_responses(
+        request,
+        expected_count=1,
+        drain=True,
+        require_content=True,
+    )

tests/fault_tolerance/cancellation/test_vllm.py

@@ -51,9 +51,8 @@
 DECODE_CANCEL_TEST_TIMEOUT_S = 900
 
 # The streaming read had no bound. STREAM_READ is the per-read socket timeout
-# between chunks; BEHAVIORAL bounds the wait for the next chunk while the
-# chunk-count goal is unmet. Neither caps total read time -- a late final chunk
-# that completes the count still counts. See read_streaming_responses.
+# between chunks; BEHAVIORAL caps total time spent reaching the chunk-count goal.
+# See read_streaming_responses.
 DECODE_CANCEL_STREAM_READ_TIMEOUT_S = 30
 DECODE_CANCEL_BEHAVIORAL_ALLOWANCE_S = 90