feat(sglang): tokens in/out with logprobs

components/src/dynamo/frontend/sglang_processor.py

@@ -41,7 +41,13 @@
     detect_force_reasoning_from_template,
     preprocess_chat_request,
 )
-from .utils import PreprocessError, extract_mm_urls, random_uuid, worker_warmup
+from .utils import (
+    PreprocessError,
+    extract_mm_urls,
+    nvext_extra_field_requested,
+    random_uuid,
+    worker_warmup,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -197,13 +203,17 @@ def _build_dynamo_preproc(
     max_tokens = request.get("max_completion_tokens") or request.get("max_tokens")
 
     stop = request.get("stop")
+    stop_token_ids = request.get("stop_token_ids", [])
     if isinstance(stop, str):
         stop = [stop]
+    elif isinstance(stop, list) and all(
+        isinstance(item, int) and not isinstance(item, bool) for item in stop
+    ):
+        stop_token_ids = [*stop_token_ids, *stop]
+        stop = []
     elif stop is None:
         stop = []
 
-    stop_token_ids = request.get("stop_token_ids", [])
-
     # Handle logprobs
     logprobs_val = None
     logprobs = request.get("logprobs")
@@ -246,6 +256,7 @@ def _build_dynamo_preproc(
             # (e.g. <|tool_call|>) to detect calls. Mirrors the
             # post-processor's _skip_special_tokens logic.
             "skip_special_tokens": tool_call_parser is None,
+            "return_tokens_as_token_ids": request.get("return_tokens_as_token_ids"),
         },
         "eos_token_ids": [eos_token_id] if eos_token_id is not None else [],
         "annotations": [],
@@ -519,6 +530,7 @@ async def _generate_and_stream(
                 new_ids = engine_response["token_ids"]
                 raw_finish = engine_response.get("finish_reason")
                 finish_reason = _map_finish_reason(raw_finish)
+                stop_reason = engine_response.get("stop_reason")
 
                 if usage := engine_response.get("completion_usage"):
                     pending_usage = usage
@@ -554,6 +566,10 @@ async def _generate_and_stream(
                         }
                         if pending_usage:
                             dynamo_out["usage"] = pending_usage
+                        if stop_reason is not None and nvext_extra_field_requested(
+                            request, "stop_reason"
+                        ):
+                            dynamo_out["nvext"] = {"stop_reason": stop_reason}
 
                         yield dynamo_out
 

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

@@ -10,6 +10,7 @@
 """
 
 
+import asyncio
 import json
 import sys
 import types
@@ -34,12 +35,18 @@
 )
 from dynamo.frontend.sglang_processor import (
     SglangPreprocessWorkerResult,
+    SglangProcessor,
     _build_dynamo_preproc,
     _init_worker,
     _map_finish_reason,
     _runtime_config_parser_name,
 )
-from dynamo.frontend.utils import PreprocessError, random_call_id, random_uuid
+from dynamo.frontend.utils import (
+    PreprocessError,
+    nvext_extra_field_requested,
+    random_call_id,
+    random_uuid,
+)
 
 # Needs sglang packages (gpu_1 container).  No need for parallel marker.
 pytestmark = [
@@ -252,6 +259,62 @@ def test_model_name_and_token_ids(self):
         assert result["model"] == "my-model"
         assert result["token_ids"] == [10, 20, 30]
 
+    def test_stop_token_id_array_maps_to_stop_token_ids(self):
+        """Integer stop arrays are token-id stops, not string stops."""
+        result = _build_dynamo_preproc(
+            {"model": "test", "stop": [32, 34]},
+            [1],
+            "test",
+            None,
+        )
+
+        assert result["stop_conditions"]["stop"] == []
+        assert result["stop_conditions"]["stop_token_ids"] == [32, 34]
+
+    def test_string_stops_remain_string_stops(self):
+        """String stops are forwarded as string stops."""
+        result = _build_dynamo_preproc(
+            {"model": "test", "stop": " The"},
+            [1],
+            "test",
+            None,
+        )
+
+        assert result["stop_conditions"]["stop"] == [" The"]
+        assert result["stop_conditions"]["stop_token_ids"] == []
+
+        result = _build_dynamo_preproc(
+            {"model": "test", "stop": ["A", "B"]},
+            [1],
+            "test",
+            None,
+        )
+
+        assert result["stop_conditions"]["stop"] == ["A", "B"]
+        assert result["stop_conditions"]["stop_token_ids"] == []
+
+    def test_token_id_display_string_remains_string_stop(self):
+        """token_id:N strings are output display strings, not token-id stops."""
+        result = _build_dynamo_preproc(
+            {"model": "test", "stop": "token_id:576"},
+            [1],
+            "test",
+            None,
+        )
+
+        assert result["stop_conditions"]["stop"] == ["token_id:576"]
+        assert result["stop_conditions"]["stop_token_ids"] == []
+
+        result = _build_dynamo_preproc(
+            {"model": "test", "stop": ["token_id:576"]},
+            [1],
+            "test",
+            None,
+        )
+
+        assert result["stop_conditions"]["stop"] == ["token_id:576"]
+        assert result["stop_conditions"]["stop_token_ids"] == []
+
 
 # ---------------------------------------------------------------------------
 # _map_finish_reason
@@ -1567,6 +1630,58 @@ def test_finish_reason_only(self, tokenizer):
         assert choice is not None
         assert choice["finish_reason"] == "stop"
 
+    def test_stop_reason_not_emitted_on_choice(self, tokenizer):
+        """Backend stop_reason is not part of the OpenAI choice shape."""
+        post = SglangStreamingPostProcessor(
+            tokenizer=tokenizer, tool_call_parser=None, reasoning_parser=None
+        )
+
+        choice = post.process_output(
+            {"token_ids": [], "finish_reason": "stop", "stop_reason": "END"}
+        )
+
+        assert choice is not None
+        assert "stop_reason" not in choice
+
+    def test_stop_reason_emits_in_nvext_when_requested(self, tokenizer):
+        """Frontend emits backend stop_reason under nvext when requested."""
+
+        class FakeRouter:
+            async def generate(self, *args, **kwargs):
+                yield {
+                    "token_ids": [],
+                    "finish_reason": "stop",
+                    "stop_reason": "END",
+                }
+
+        async def collect():
+            processor = SglangProcessor(
+                tokenizer=tokenizer,
+                router=FakeRouter(),
+                tool_call_parser_name=None,
+                reasoning_parser_name=None,
+                eos_token_id=None,
+            )
+            post = SglangStreamingPostProcessor(
+                tokenizer=tokenizer, tool_call_parser=None, reasoning_parser=None
+            )
+            request = {
+                "model": "test-model",
+                "nvext": {"extra_fields": ["stop_reason"]},
+            }
+            return [
+                item
+                async for item in processor._generate_and_stream(
+                    "req-stop", request, {}, [], post
+                )
+            ]
+
+        items = asyncio.run(collect())
+
+        assert len(items) == 1
+        assert items[0]["nvext"]["stop_reason"] == "END"
+        assert "stop_reason" not in items[0]["choices"][0]
+
     def test_lookback_trimming(self, tokenizer):
         """Verify _all_token_ids doesn't grow unbounded."""
         post = SglangStreamingPostProcessor(
@@ -1684,6 +1799,13 @@ def test_preprocess_error(self):  # FRONTEND.8
         err = PreprocessError("n=2 unsupported")
         assert "n=2" in str(err)
 
+    def test_nvext_extra_field_requested(self):
+        assert nvext_extra_field_requested(
+            {"nvext": {"extra_fields": ["stop_reason"]}}, "stop_reason"
+        )
+        assert not nvext_extra_field_requested({"nvext": {}}, "stop_reason")
+        assert not nvext_extra_field_requested({}, "stop_reason")
+
 
 # ---------------------------------------------------------------------------
 # SglangPreprocessWorkerResult picklability

components/src/dynamo/frontend/utils.py

@@ -20,6 +20,15 @@ def random_call_id() -> str:
     return f"call_{uuid.uuid4().int & _MASK_64_BITS:016x}"
 
 
+def nvext_extra_field_requested(request: dict[str, Any], field: str) -> bool:
+    """Return whether a request opted into a response nvext field."""
+    nvext = request.get("nvext")
+    if not isinstance(nvext, dict):
+        return False
+    extra_fields = nvext.get("extra_fields")
+    return isinstance(extra_fields, list) and field in extra_fields
+
+
 def worker_warmup() -> bool:
     """Dummy task to ensure a ProcessPoolExecutor worker is fully initialized."""
     return True

components/src/dynamo/sglang/CLAUDE.md

@@ -228,6 +228,12 @@ absolute sequence position where logprob computation starts: `-1` (default) = ou
 only (`len(prompt) - 1`), `0` = from prompt start. We set it to 0 when `prompt_logprobs`
 is requested.
 
+**Top-logprobs gate**: `logprobs >= 1` (or `prompt_logprobs >= 1`) raises `ValueError`
+by default. SGLang's tokenizer manager detokenizes top-k tokens per-position serially,
+causing severe latency degradation (O(N) per generated token). Callers must use
+`logprobs=0` for chosen-token-only logprobs. Set `DYN_SGL_ALLOW_TOP_LOGPROBS=1` to
+override once upstream batches `detokenize_top_logprobs_tokens`.
+
 **Streaming behavior** (`_extract_logprobs`):
 
 Dynamo forces `stream_output=True` (args.py:374), making `output_ids` disjoint per chunk.

components/src/dynamo/sglang/protocol.py

@@ -19,6 +19,7 @@
 class StopConditions(BaseModel):
     max_tokens: Optional[int] = None
     stop: Optional[List[str]] = None
+    stop_token_ids: Optional[List[TokenIdType]] = None
     stop_token_ids_hidden: Optional[List[TokenIdType]] = None
     min_tokens: Optional[int] = None
     ignore_eos: Optional[bool] = None