fix(sglang): [cherry-pick 1.5.1] support min_tokens on tokenizer-free decode workers (#14276, #14378)

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

@@ -1134,15 +1134,30 @@ def _get_bootstrap_info(engine: sgl.Engine) -> Tuple[str, int]:
 
         return bootstrap_host, bootstrap_port
 
+    def _abort_requests(self, request_ids: set[str], context: Context) -> None:
+        if not request_ids:
+            return
+        if not getattr(self.engine, "tokenizer_manager", None):
+            logging.error("SGLang tokenizer_manager not found for abort requests")
+            return
+        for request_id in request_ids:
+            self.engine.tokenizer_manager.abort_request(rid=request_id, abort_all=False)
+        request_ids.clear()
+        logging.info("Aborted Request ID: %s", context.id())
+
     async def _handle_cancellation(
-        self, request_id_future: asyncio.Future, context: Context
+        self,
+        request_id_future: asyncio.Future,
+        context: Context,
+        request_ids: set[str] | None = None,
     ):
         """Background task to handle cancellation and shutdown by monitoring both signals.
 
         Args:
             request_id_future: Future that will be set with the SGLang request ID
                               when the first response arrives.
             context: Context object for cancellation handling.
+            request_ids: Live set of unfinished parallel-choice IDs, when available.
 
         Raises:
             EngineShutdown: If shutdown event was triggered.
@@ -1188,22 +1203,9 @@ async def _handle_cancellation(
                 f"Cancellation or shutdown signal received for SGLang Request ID {sglang_request_id}, Context: {context.id()}"
             )
 
-            # Call abort_request on the tokenizer_manager through the engine
-            if (
-                hasattr(self.engine, "tokenizer_manager")
-                and self.engine.tokenizer_manager
-            ):
-                logging.info(
-                    f"Calling SGLang abort_request for Request ID {sglang_request_id}"
-                )
-                self.engine.tokenizer_manager.abort_request(
-                    rid=sglang_request_id, abort_all=False
-                )
-                logging.info(f"Aborted Request ID: {context.id()}")
-            else:
-                logging.error(
-                    f"SGLang tokenizer_manager not found for abort request: {context.id()}"
-                )
+            self._abort_requests(
+                request_ids if request_ids is not None else {sglang_request_id}, context
+            )
 
             # Check which event triggered and raise EngineShutdown if shutdown
             if shutdown_task and shutdown_task in done:
@@ -1233,7 +1235,10 @@ async def _handle_cancellation(
 
     @asynccontextmanager
     async def _cancellation_monitor(
-        self, request_id_future: asyncio.Future, context: Context
+        self,
+        request_id_future: asyncio.Future,
+        context: Context,
+        request_ids: set[str] | None = None,
     ) -> AsyncGenerator[asyncio.Task, None]:
         """
         Context manager for monitoring request cancellation and shutdown.
@@ -1246,6 +1251,7 @@ async def _cancellation_monitor(
             request_id_future: Future that will be set with the SGLang request ID
                               when the first response arrives.
             context: Context object for cancellation handling
+            request_ids: Live set of unfinished parallel-choice IDs to abort on exit.
 
         Yields:
             asyncio.Task: The cancellation monitoring task being managed
@@ -1254,12 +1260,15 @@ async def _cancellation_monitor(
 
         # Start the cancellation monitoring task
         cancellation_task = asyncio.create_task(
-            self._handle_cancellation(request_id_future, context)
+            self._handle_cancellation(request_id_future, context, request_ids)
         )
 
         try:
             yield cancellation_task
         finally:
+            # Stream closure can beat the monitor's cancellation wake-up.
+            if request_ids is not None:
+                self._abort_requests(request_ids, context)
             # Clean up the background cancellation task
             request_id = "unknown"
             if request_id_future.done() and not request_id_future.cancelled():

components/src/dynamo/sglang/request_handlers/llm/decode_handler.py

@@ -360,12 +360,25 @@ def _build_sampling_params(self, request: Dict[str, Any]) -> Dict[str, Any]:
             _plain = stop_conditions.get("stop_token_ids") or []
             _merged = list(set(_hidden).union(_plain))
             stop_token_ids = _merged if _merged else None
+            # A tokenizer-free SGLang rejects positive min_new_tokens. Let
+            # Dynamo's decoder enforce the floor and keep the engine running so
+            # it cannot stop before Dynamo reaches it.
+            min_tokens = stop_conditions.get("min_tokens")
+            hold_engine_open = bool(
+                min_tokens
+                and min_tokens > 0
+                and self.config.server_args.skip_tokenizer_init
+            )
+            if hold_engine_open:
+                stop_token_ids = None
 
             param_mapping = {
                 "n": sampling_opts.get("n"),
                 "max_new_tokens": stop_conditions.get("max_tokens"),
-                "min_new_tokens": stop_conditions.get("min_tokens"),
-                "ignore_eos": stop_conditions.get("ignore_eos"),
+                "min_new_tokens": None if hold_engine_open else min_tokens,
+                "ignore_eos": True
+                if hold_engine_open
+                else stop_conditions.get("ignore_eos"),
                 "stop_token_ids": stop_token_ids,
                 **_sampling_option_params(sampling_opts),
                 **self._get_guided_decoding_params(
@@ -691,16 +704,23 @@ async def _process_token_stream(
         """
         # Use Future pattern for request ID - will be set when first response arrives
         request_id_future: asyncio.Future[str] = asyncio.Future()
+        request_ids: set[str] = set()
         first_output_seen = False
-        async with self._cancellation_monitor(request_id_future, context):
+        async with self._cancellation_monitor(request_id_future, context, request_ids):
             async for res in stream_source:
                 meta_info = res.get("meta_info", {})
-                # Extract SGLang request ID from the first response and set the future
-                if not request_id_future.done():
-                    sglang_request_id = meta_info.get("id")
-                    if sglang_request_id:
+                sglang_request_id = meta_info.get("id")
+                if sglang_request_id:
+                    request_ids.add(sglang_request_id)
+                    if not request_id_future.done():
                         request_id_future.set_result(sglang_request_id)
                         logging.debug(f"New SGLang Request ID: {sglang_request_id}")
+                    if meta_info.get("finish_reason"):
+                        request_ids.discard(sglang_request_id)
+                if context.is_stopped():
+                    # A choice's first chunk can arrive after the monitor fired.
+                    self._abort_requests(request_ids, context)
+                    continue
 
                 # Check cancellation before yielding to allow proper cleanup.
                 # This lets SGLang proceed to the second token generation, which will
@@ -829,16 +849,22 @@ async def _process_text_stream(
 
         # Use Future pattern for request ID - will be set when first response arrives
         request_id_future: asyncio.Future[str] = asyncio.Future()
+        request_ids: set[str] = set()
         first_output_seen = False
-        async with self._cancellation_monitor(request_id_future, context):
+        async with self._cancellation_monitor(request_id_future, context, request_ids):
             async for res in stream_source:
                 meta_info = res.get("meta_info", {})
-                # Extract SGLang request ID from the first response and set the future
-                if not request_id_future.done():
-                    sglang_request_id = meta_info.get("id")
-                    if sglang_request_id:
+                sglang_request_id = meta_info.get("id")
+                if sglang_request_id:
+                    request_ids.add(sglang_request_id)
+                    if not request_id_future.done():
                         request_id_future.set_result(sglang_request_id)
                         logging.debug(f"New SGLang Request ID: {sglang_request_id}")
+                    if meta_info.get("finish_reason"):
+                        request_ids.discard(sglang_request_id)
+                if context.is_stopped():
+                    self._abort_requests(request_ids, context)
+                    continue
 
                 # Check cancellation before yielding to allow proper cleanup.
                 # This lets SGLang proceed to the second token generation, which will

components/src/dynamo/sglang/tests/test_sglang_decode_handler.py

@@ -2,8 +2,9 @@
 # SPDX-License-Identifier: Apache-2.0
 
 import asyncio
-from contextlib import asynccontextmanager
+from contextlib import aclosing, asynccontextmanager
 from types import SimpleNamespace
+from unittest.mock import Mock, call
 
 import pytest
 
@@ -239,12 +240,20 @@ def test_openai_stop_sampling_params_maps_token_id_stop_array():
     }
 
 
-def _new_decode_handler(*, use_sglang_tokenizer: bool = False, enable_rl: bool = False):
+def _new_decode_handler(
+    *,
+    use_sglang_tokenizer: bool = False,
+    skip_tokenizer_init: bool = False,
+    enable_rl: bool = False,
+):
     handler = DecodeWorkerHandler.__new__(DecodeWorkerHandler)
     handler.shutdown_event = None
     handler.use_sglang_tokenizer = use_sglang_tokenizer
     handler.config = SimpleNamespace(
-        server_args=SimpleNamespace(served_model_name="test-model"),
+        server_args=SimpleNamespace(
+            served_model_name="test-model",
+            skip_tokenizer_init=skip_tokenizer_init,
+        ),
         dynamo_args=SimpleNamespace(enable_rl=enable_rl),
     )
     handler._first_token_source = None
@@ -257,6 +266,68 @@ async def no_cancellation_monitor(*args, **kwargs):
     return handler
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "processor_name", ["_process_token_stream", "_process_text_stream"]
+)
+@pytest.mark.parametrize("cancel_from_client", [False, True], ids=["close", "cancel"])
+async def test_stream_cancellation_aborts_unfinished_choices(
+    processor_name, cancel_from_client, caplog
+):
+    handler = _new_decode_handler()
+    # Exercise the real monitor, including closure before its task can run.
+    del handler._cancellation_monitor
+    cancellation = asyncio.get_running_loop().create_future()
+    first_abort = asyncio.Event()
+    abort_request = Mock(side_effect=lambda **_kwargs: first_abort.set())
+    handler.engine = SimpleNamespace(
+        tokenizer_manager=SimpleNamespace(abort_request=abort_request)
+    )
+    context = SimpleNamespace(
+        id=lambda: "context-request-id",
+        is_stopped=cancellation.done,
+        async_killed_or_stopped=lambda: cancellation,
+        notify_first_token=lambda: None,
+    )
+    caplog.set_level("INFO")
+
+    async def stream():
+        for index in range(4):
+            if index == 3:
+                # A new choice can first appear after the monitor has aborted
+                # the choices it already knows about.
+                await first_abort.wait()
+            yield {
+                "index": index,
+                "text": "hello",
+                "output_ids": [11],
+                "meta_info": {
+                    "id": f"choice-{index}",
+                    "finish_reason": {"type": "length"} if index == 2 else None,
+                },
+            }
+
+    async with aclosing(getattr(handler, processor_name)(stream(), context)) as output:
+        for _ in range(3):
+            await anext(output)
+        abort_request.assert_not_called()
+        if cancel_from_client:
+            cancellation.set_result(None)
+            with pytest.raises(StopAsyncIteration):
+                await asyncio.wait_for(anext(output), timeout=5)
+
+    expected_ids = ["choice-0", "choice-1"]
+    if cancel_from_client:
+        expected_ids.append("choice-3")
+    abort_request.assert_has_calls(
+        [call(rid=request_id, abort_all=False) for request_id in expected_ids],
+        any_order=True,
+    )
+    assert abort_request.call_count == len(expected_ids)
+    # Preserve the context-ID log consumed by the GPU cancellation tests.
+    assert "Aborted Request ID: context-request-id" in caplog.messages
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
     "processor_name", ["_process_token_stream", "_process_text_stream"]
@@ -265,7 +336,7 @@ async def test_shutdown_abort_chunk_raises_engine_shutdown(processor_name):
     handler = _new_decode_handler()
     handler.shutdown_event = asyncio.Event()
     handler.shutdown_event.set()
-    context = SimpleNamespace(id=lambda: "request-id")
+    context = SimpleNamespace(id=lambda: "request-id", is_stopped=lambda: False)
 
     async def stream():
         yield {
@@ -747,23 +818,55 @@ def test_build_sampling_params_maps_guided_decoding_to_regex():
     assert sampling_params["regex"] == r"\{\}"
 
 
-def test_build_sampling_params_maps_min_tokens_for_token_requests():
-    handler = _new_decode_handler(use_sglang_tokenizer=False)
+def test_build_sampling_params_holds_tokenizer_free_engine_open_for_min_tokens():
+    handler = _new_decode_handler(
+        use_sglang_tokenizer=False,
+        skip_tokenizer_init=True,
+    )
 
     sampling_params = handler._build_sampling_params(
         {
             "sampling_options": {},
             "stop_conditions": {
                 "max_tokens": 64,
-                "min_tokens": 64,
-                "ignore_eos": True,
+                "min_tokens": 8,
+                "stop_token_ids": [7],
+                "stop_token_ids_hidden": [151643],
             },
         }
     )
 
-    assert sampling_params["min_new_tokens"] == 64
-    assert sampling_params["max_new_tokens"] == 64
+    assert "min_new_tokens" not in sampling_params
+    assert "stop_token_ids" not in sampling_params
     assert sampling_params["ignore_eos"] is True
+    assert sampling_params["max_new_tokens"] == 64
+
+    from sglang.srt.sampling.sampling_params import SamplingParams
+
+    SamplingParams(**sampling_params).normalize(tokenizer=None)
+
+
+def test_build_sampling_params_forwards_min_tokens_when_engine_has_tokenizer():
+    handler = _new_decode_handler(
+        use_sglang_tokenizer=False,
+        skip_tokenizer_init=False,
+    )
+
+    sampling_params = handler._build_sampling_params(
+        {
+            "sampling_options": {},
+            "stop_conditions": {
+                "max_tokens": 64,
+                "min_tokens": 8,
+                "stop_token_ids": [7],
+            },
+        }
+    )
+
+    assert sampling_params["min_new_tokens"] == 8
+    assert sampling_params["stop_token_ids"] == [7]
+    assert "ignore_eos" not in sampling_params
+    assert sampling_params["max_new_tokens"] == 64
 
 
 def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests():
@@ -778,7 +881,10 @@ def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests():
 
 
 def test_build_sampling_params_omits_min_new_tokens_when_min_tokens_absent():
-    handler = _new_decode_handler(use_sglang_tokenizer=False)
+    handler = _new_decode_handler(
+        use_sglang_tokenizer=False,
+        skip_tokenizer_init=True,
+    )
 
     sampling_params = handler._build_sampling_params(
         {"sampling_options": {}, "stop_conditions": {"max_tokens": 8}}
@@ -788,7 +894,10 @@ def test_build_sampling_params_omits_min_new_tokens_when_min_tokens_absent():
 
 
 def test_build_sampling_params_forwards_explicit_zero_min_tokens():
-    handler = _new_decode_handler(use_sglang_tokenizer=False)
+    handler = _new_decode_handler(
+        use_sglang_tokenizer=False,
+        skip_tokenizer_init=True,
+    )
 
     sampling_params = handler._build_sampling_params(
         {"sampling_options": {}, "stop_conditions": {"max_tokens": 8, "min_tokens": 0}}

container/deps/requirements.test.txt

@@ -38,6 +38,8 @@ mistral-common>=1.11.2
 mypy==1.18.2
 # For NATS object store verification in router tests
 nats-py==2.12.0
+# Router benchmark tests import graph synthesis during whole-tree collection.
+networkx
 openai==2.32.0  # for Fault Tolerance migration tests
 pandas-stubs>=1.3.0
 psutil<=7.0.0  # System package, may vary by platform (was >=5.0.0)

deploy/operator/docs/footer.md

@@ -344,7 +344,6 @@ For users who want to understand the implementation details or contribute to the
 - **Checkpoint / Restore**:
   - [`internal/checkpoint/podspec.go`](https://github.com/ai-dynamo/dynamo/blob/main/deploy/operator/internal/checkpoint/podspec.go) - Checkpoint env var injection and volume setup
   - [`internal/checkpoint/resolve.go`](https://github.com/ai-dynamo/dynamo/blob/main/deploy/operator/internal/checkpoint/resolve.go) - Checkpoint resolution logic
-  - [`internal/checkpoint/resource.go`](https://github.com/ai-dynamo/dynamo/blob/main/deploy/operator/internal/checkpoint/resource.go) - Checkpoint resource management
 - **Constants & Annotations**: [`internal/consts/consts.go`](https://github.com/ai-dynamo/dynamo/blob/main/deploy/operator/internal/consts/consts.go) - Defines annotation keys and other constants
 
 ## Notes