fix(sglang): pass reasoning gate for guided decoding

Bridge Dynamo's guided-decoding reasoning state into SGLang by forwarding require_reasoning from the frontend/preprocessor and setting GenerateReqInput.require_reasoning before SGLang initializes its grammar backend.

Mirror Dynamo reasoning-parser names into SGLang parser names, cover the unified SglangLLMEngine path plus legacy prefill/decode handlers, and add request-gating coverage for thinking opt-out cases.

Signed-off-by: Yuting Wu (DLAlgo) <yutwu@nvidia.com>

Author: nvyutwu

components/src/dynamo/frontend/sglang_processor.py

@@ -158,6 +158,7 @@ def _preprocess_worker(
         eos_token_id,
         pre.guided_decoding,
         pre.tool_call_parser,
+        require_reasoning=pre.force_reasoning,
     )
 
     effective_reasoning_parser_name = (
@@ -180,6 +181,7 @@ def _build_dynamo_preproc(
     eos_token_id: int | None,
     guided_decoding: dict[str, Any] | None = None,
     tool_call_parser: ToolCallParserType | None = None,
+    require_reasoning: bool = False,
 ) -> dict[str, Any]:
     """Build the Dynamo preprocessed request dict from request fields."""
     max_tokens = request.get("max_completion_tokens") or request.get("max_tokens")
@@ -250,6 +252,11 @@ def _build_dynamo_preproc(
     if mm_data:
         preproc["multi_modal_data"] = mm_data
 
+    if require_reasoning:
+        extra_args = dict(preproc.get("extra_args") or {})
+        extra_args["require_reasoning"] = True
+        preproc["extra_args"] = extra_args
+
     return preproc
 
 
@@ -368,6 +375,7 @@ async def _generator_inner(
                 self.eos_token_id,
                 pre.guided_decoding,
                 pre.tool_call_parser,
+                require_reasoning=pre.force_reasoning,
             )
         except InvalidArgument:
             raise

components/src/dynamo/frontend/tests/test_sglang_processor_unit.py

@@ -112,6 +112,18 @@ def test_top_k_positive_preserved(self):
         )
         assert result["sampling_options"]["top_k"] == 50
 
+    def test_require_reasoning_forwards_backend_extra_args(self):
+        """Reasoning-gated guided decoding must reach the SGLang backend."""
+        result = _build_dynamo_preproc(
+            {"model": "test"},
+            prompt_token_ids=[1],
+            model_name="test",
+            eos_token_id=None,
+            require_reasoning=True,
+        )
+        assert result["extra_args"]["require_reasoning"] is True
+        assert "prompt_injected_reasoning" not in result["extra_args"]
+
     def test_sampling_options_from_request(self):
         """All sampling fields are projected from request."""
         request = {

components/src/dynamo/sglang/args.py

@@ -30,6 +30,24 @@
 configure_dynamo_logging()
 
 
+_DYN_TO_SGLANG_REASONING_PARSER = {
+    "deepseek_r1": "deepseek-r1",
+    "deepseek_v3": "deepseek-v3",
+    "deepseek_v3_1": "deepseek-v3",
+    "deepseek_v3_2": "deepseek-v3",
+    "deepseek_v4": "deepseek-v4",
+    "deepseekv4": "deepseek-v4",
+    "gemma-4": "gemma4",
+    "gpt_oss": "gpt-oss",
+    "kimi_k25": "kimi_k2",
+    "minimax_append_think": "minimax-append-think",
+    "nemotron_nano": "nemotron_3",
+    "nemotron3": "nemotron_3",
+    "nemotron_v3": "nemotron_3",
+    "nemotron_deci": "glm45",
+}
+
+
 class DynamoConfig(DynamoRuntimeConfig, DynamoSGLangConfig):
     """Combined configuration container for SGLang server and Dynamo args."""
 
@@ -79,12 +97,44 @@ def _preprocess_for_encode_config(
 def _validate_parser_flags(
     sglang_val: Optional[str], dynamo_val: Optional[str], name: str
 ) -> None:
-    """Validate that --{name} (SGLang) and --dyn-{name} (Dynamo) are not both set."""
+    """Validate parser flag combinations."""
+    if name == "reasoning-parser" and sglang_val and dynamo_val:
+        expected_sglang_val = _DYN_TO_SGLANG_REASONING_PARSER.get(
+            dynamo_val, dynamo_val
+        )
+        if sglang_val != expected_sglang_val:
+            logging.error(
+                "Cannot use different --reasoning-parser (%s) and "
+                "--dyn-reasoning-parser (%s) values. Expected SGLang parser "
+                "%s for Dynamo parser %s.",
+                sglang_val,
+                dynamo_val,
+                expected_sglang_val,
+                dynamo_val,
+            )
+            sys.exit(1)
+        return
     if sglang_val and dynamo_val:
         logging.error(f"Cannot use both --{name} and --dyn-{name}.")
         sys.exit(1)
 
 
+def _mirror_dyn_reasoning_parser_to_sglang(
+    parsed_args: argparse.Namespace, dynamo_config: DynamoSGLangConfig
+) -> None:
+    """Use Dynamo's reasoning parser to activate SGLang's reasoner gate."""
+    dyn_parser = dynamo_config.dyn_reasoning_parser
+    if dyn_parser and not getattr(parsed_args, "reasoning_parser", None):
+        sglang_parser = _DYN_TO_SGLANG_REASONING_PARSER.get(dyn_parser, dyn_parser)
+        parsed_args.reasoning_parser = sglang_parser
+        logging.info(
+            "Mirroring --dyn-reasoning-parser=%s to SGLang --reasoning-parser=%s "
+            "so guided decoding is gated until the reasoning end token.",
+            dyn_parser,
+            sglang_parser,
+        )
+
+
 def _has_cli_flag(args: list[str], flag: str) -> bool:
     """Return True when a CLI flag is present in '--flag val' or '--flag=val' form."""
     return any(arg == flag or arg.startswith(f"{flag}=") for arg in args)
@@ -295,6 +345,7 @@ async def parse_args(args: list[str]) -> Config:
         dynamo_config.dyn_reasoning_parser,
         "reasoning-parser",
     )
+    _mirror_dyn_reasoning_parser_to_sglang(parsed_args, dynamo_config)
 
     if dynamo_config.custom_jinja_template and dynamo_config.use_sglang_tokenizer:
         logging.error(

components/src/dynamo/sglang/llm_engine.py

@@ -59,6 +59,10 @@
     runtime_capacity,
 )
 from dynamo.sglang.publisher import format_zmq_endpoint
+from dynamo.sglang.reasoning import (
+    install_require_reasoning_proxy,
+    require_reasoning_context,
+)
 
 if TYPE_CHECKING:
     from dynamo._core.backend import EngineMetrics  # type: ignore[import-not-found]
@@ -149,6 +153,7 @@ async def start(self, worker_id: int) -> EngineConfig:
         del worker_id  # SGLang bootstrap uses host/port/room triples
 
         self.engine = sgl.Engine(server_args=self.server_args)
+        install_require_reasoning_proxy(self.engine)
 
         tokenizer = (
             self.engine.tokenizer_manager.tokenizer
@@ -289,19 +294,22 @@ async def generate(
             "SGLang",
         )
 
-        stream = await self.engine.async_generate(
-            **input_param,
-            sampling_params=sampling_params,
-            stream=True,
-            rid=context.trace_id,
-            data_parallel_rank=sgl_dp_rank,
-            **telemetry.engine_trace_kwargs(
-                context,
-                kwarg_name="external_trace_header",
-                enabled=self.enable_trace,
-            ),
-            **bootstrap_kwargs,
-        )
+        with require_reasoning_context(
+            self._has_reasoning_parser(), dict(request), input_param
+        ):
+            stream = await self.engine.async_generate(
+                **input_param,
+                sampling_params=sampling_params,
+                stream=True,
+                rid=context.trace_id,
+                data_parallel_rank=sgl_dp_rank,
+                **telemetry.engine_trace_kwargs(
+                    context,
+                    kwarg_name="external_trace_header",
+                    enabled=self.enable_trace,
+                ),
+                **bootstrap_kwargs,
+            )
 
         # ORDER MATTERS: async_generate must register the room (the await
         # above) before we yield the bootstrap chunk — otherwise the
@@ -719,3 +727,9 @@ def _get_input_param(self, request: GenerateRequest) -> dict:
         return {
             "prompt" if isinstance(request_input, str) else "input_ids": request_input
         }
+
+    def _has_reasoning_parser(self) -> bool:
+        return bool(
+            getattr(self.server_args, "reasoning_parser", None)
+            or getattr(self.dynamo_args, "dyn_reasoning_parser", None)
+        )

components/src/dynamo/sglang/reasoning.py

@@ -0,0 +1,65 @@
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+import contextvars
+import functools
+import threading
+from contextlib import contextmanager
+from typing import Any, Dict
+
+_DYN_REQUIRE_REASONING_CV: contextvars.ContextVar[bool] = contextvars.ContextVar(
+    "dynamo_sglang_require_reasoning", default=False
+)
+_REQUIRE_REASONING_PROXY_LOCK = threading.Lock()
+
+
+def install_require_reasoning_proxy(engine: Any) -> None:
+    """Set SGLang GenerateReqInput.require_reasoning from a per-request context."""
+    tm = getattr(engine, "tokenizer_manager", None)
+    if tm is None or getattr(tm, "_dynamo_require_reasoning_wrapped", False):
+        return
+
+    with _REQUIRE_REASONING_PROXY_LOCK:
+        if getattr(tm, "_dynamo_require_reasoning_wrapped", False):
+            return
+
+        original = tm.generate_request
+
+        @functools.wraps(original)
+        def _wrapped(obj, request):
+            if _DYN_REQUIRE_REASONING_CV.get():
+                obj.require_reasoning = True
+            return original(obj, request)
+
+        tm._dynamo_require_reasoning_wrapped = True  # type: ignore[attr-defined]
+        tm.generate_request = _wrapped  # type: ignore[assignment]
+
+
+def request_requires_reasoning(
+    has_reasoning_parser: bool, request: Dict[str, Any], input_param: Dict[str, Any]
+) -> bool:
+    if not has_reasoning_parser:
+        return False
+
+    extra_args = request.get("extra_args") or {}
+    if isinstance(extra_args, dict) and (
+        extra_args.get("require_reasoning")
+        or extra_args.get("prompt_injected_reasoning")
+    ):
+        return True
+
+    prompt = input_param.get("prompt")
+    return isinstance(prompt, str) and prompt.rstrip().endswith("<think>")
+
+
+@contextmanager
+def require_reasoning_context(
+    has_reasoning_parser: bool, request: Dict[str, Any], input_param: Dict[str, Any]
+):
+    token = _DYN_REQUIRE_REASONING_CV.set(
+        request_requires_reasoning(has_reasoning_parser, request, input_param)
+    )
+    try:
+        yield
+    finally:
+        _DYN_REQUIRE_REASONING_CV.reset(token)

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

@@ -44,6 +44,10 @@
 from dynamo.runtime import DistributedRuntime
 from dynamo.sglang.args import Config
 from dynamo.sglang.publisher import DynamoSglangPublisher
+from dynamo.sglang.reasoning import (
+    install_require_reasoning_proxy,
+    require_reasoning_context,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -720,6 +724,7 @@ def __init__(
         self.enable_trace = getattr(config.server_args, "enable_trace", False)
 
         if engine is not None:
+            install_require_reasoning_proxy(engine)
             self.input_param_manager = InputParamManager(
                 self.engine.tokenizer_manager.tokenizer
                 if self.use_sglang_tokenizer
@@ -1059,6 +1064,19 @@ def _get_input_param(self, request: Dict[str, Any]) -> Dict[str, Any]:
             "prompt" if isinstance(request_input, str) else "input_ids": request_input
         }
 
+    def _has_reasoning_parser(self) -> bool:
+        return bool(
+            getattr(self.config.server_args, "reasoning_parser", None)
+            or getattr(self.config.dynamo_args, "dyn_reasoning_parser", None)
+        )
+
+    def _require_reasoning_context(
+        self, request: Dict[str, Any], input_param: Dict[str, Any]
+    ):
+        return require_reasoning_context(
+            self._has_reasoning_parser(), request, input_param
+        )
+
     def _session_kwargs(self, request: Dict[str, Any]) -> Dict[str, Any]:
         if not getattr(self.config.server_args, "enable_streaming_session", False):
             return {}

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

@@ -466,22 +466,23 @@ async def generate(
             routing = request.get("routing") or {}
             dp_rank = routing.get("dp_rank")
 
-            decode = await self.engine.async_generate(
-                **input_param,
-                sampling_params=sampling_params,
-                stream=True,
-                **self._routed_experts_kwargs,
-                bootstrap_host=bootstrap_info["bootstrap_host"],
-                bootstrap_port=bootstrap_info["bootstrap_port"],
-                bootstrap_room=bootstrap_info["bootstrap_room"],
-                external_trace_header=trace_header,
-                rid=trace_id,
-                data_parallel_rank=dp_rank,
-                **self._session_kwargs(request),
-                lora_path=lora_path,
-                **logprob_kwargs,
-                **self._priority_kwargs(priority),
-            )
+            with self._require_reasoning_context(request, input_param):
+                decode = await self.engine.async_generate(
+                    **input_param,
+                    sampling_params=sampling_params,
+                    stream=True,
+                    **self._routed_experts_kwargs,
+                    bootstrap_host=bootstrap_info["bootstrap_host"],
+                    bootstrap_port=bootstrap_info["bootstrap_port"],
+                    bootstrap_room=bootstrap_info["bootstrap_room"],
+                    external_trace_header=trace_header,
+                    rid=trace_id,
+                    data_parallel_rank=dp_rank,
+                    **self._session_kwargs(request),
+                    lora_path=lora_path,
+                    **logprob_kwargs,
+                    **self._priority_kwargs(priority),
+                )
 
             if not self.use_sglang_tokenizer:
                 async for out in self._process_token_stream(
@@ -525,21 +526,22 @@ async def generate(
             routing = request.get("routing") or {}
             dp_rank = routing.get("dp_rank")
 
-            agg = await self.engine.async_generate(
-                **input_param,
-                image_data=image_data,
-                video_data=video_data,
-                sampling_params=sampling_params,
-                stream=True,
-                **self._routed_experts_kwargs,
-                external_trace_header=trace_header,
-                rid=trace_id,
-                data_parallel_rank=dp_rank,
-                **self._session_kwargs(request),
-                lora_path=lora_path,
-                **logprob_kwargs,
-                **self._priority_kwargs(priority),
-            )
+            with self._require_reasoning_context(request, input_param):
+                agg = await self.engine.async_generate(
+                    **input_param,
+                    image_data=image_data,
+                    video_data=video_data,
+                    sampling_params=sampling_params,
+                    stream=True,
+                    **self._routed_experts_kwargs,
+                    external_trace_header=trace_header,
+                    rid=trace_id,
+                    data_parallel_rank=dp_rank,
+                    **self._session_kwargs(request),
+                    lora_path=lora_path,
+                    **logprob_kwargs,
+                    **self._priority_kwargs(priority),
+                )
             if not self.use_sglang_tokenizer:
                 async for out in self._process_token_stream(
                     agg,