From 6e7df975fc7ec19219b012f770a1e8188fab5319 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 3 Sep 2026 17:54:05 +0000 Subject: [PATCH 01/15] fix(sglang): stop sending min_new_tokens to a tokenizer-free engine SGLang refuses `min_new_tokens` when the engine was started without a tokenizer, because it needs the tokenizer's EOS id to apply the floor. Dynamo forwarded the parameter anyway from three lowering sites, so any `min_tokens` request against a tokenizer-free SGLang worker was rejected. Dynamo does not need the engine for this. The `Decoder` in `lib/llm/src/backend.rs` already holds the same `min_tokens` and the same stop-token set the preprocessor resolved, and suppresses every stop trigger until the floor is reached. What it cannot do is manufacture tokens the engine never produced, so when a floor is requested the engine is now held open for the request -- it receives no stop ids and is told to ignore EOS -- and termination becomes entirely Dynamo's. The engine may generate tokens Dynamo discards; that is bounded by `max_tokens`. The branch where the engine owns its tokenizer is unchanged: it still receives `min_new_tokens` and enforces the floor natively. Signed-off-by: svc-glamr@nvidia.com --- .../request_handlers/llm/decode_handler.py | 16 +++- .../multimodal/worker_handler.py | 36 +++++++-- .../tests/test_sglang_decode_handler.py | 57 +++++++++---- lib/sidecar/sglang/src/protocol.rs | 80 ++++++++++++++++--- 4 files changed, 154 insertions(+), 35 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index e7b18053d97b..69acc82d75d8 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -434,11 +434,23 @@ 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 + # min_new_tokens is never sent on this branch. SGLang rejects it when the + # engine has no tokenizer, which is exactly what this branch means, and + # Dynamo already enforces the floor: the Rust decoder holds the same + # min_tokens and the same stop-token set, and suppresses every stop until + # the floor is reached. It cannot manufacture tokens the engine never + # produced, so when a floor is requested the engine is held open instead — + # no stop ids, ignore_eos set — and termination becomes entirely Dynamo's. + # The engine may then generate tokens Dynamo discards, bounded by max_tokens. + hold_engine_open = (stop_conditions.get("min_tokens") or 0) > 0 + 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"), + "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( diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index 1a15cc59dba6..c73d28825264 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -75,8 +75,16 @@ class SglangUtils: """General SGLang utilities (not multimodal-specific)""" @staticmethod - def build_sampling_params(request: SglangMultimodalRequest) -> dict: - """Build sampling parameters for SGLang engine (generic functionality)""" + def build_sampling_params( + request: SglangMultimodalRequest, *, use_sglang_tokenizer: bool + ) -> dict: + """Build sampling parameters for SGLang engine (generic functionality) + + Args: + request: Pre-tokenized multimodal request. + use_sglang_tokenizer: Whether the engine owns a tokenizer. When it does not, + min_tokens is enforced by Dynamo rather than forwarded to the engine. + """ sampling_params = {} # Extract sampling options from request @@ -93,10 +101,20 @@ def build_sampling_params(request: SglangMultimodalRequest) -> dict: sampling_params["n"] = sampling_options.n if stop_conditions.max_tokens: sampling_params["max_new_tokens"] = stop_conditions.max_tokens + # min_new_tokens only goes to an engine that owns a tokenizer; SGLang rejects it + # otherwise. Dynamo already enforces the floor -- the Rust decoder holds the same + # min_tokens and the same stop-token set and suppresses every stop until the floor + # is reached -- but it cannot manufacture tokens the engine never produced, so a + # tokenizer-free engine is held open with ignore_eos and left to Dynamo to stop. + # The engine may then generate tokens Dynamo discards, bounded by max_tokens. + hold_engine_open = False if stop_conditions.min_tokens: - sampling_params["min_new_tokens"] = stop_conditions.min_tokens - if stop_conditions.ignore_eos: - sampling_params["ignore_eos"] = stop_conditions.ignore_eos + if use_sglang_tokenizer: + sampling_params["min_new_tokens"] = stop_conditions.min_tokens + else: + hold_engine_open = True + if hold_engine_open or stop_conditions.ignore_eos: + sampling_params["ignore_eos"] = True logger.debug(f"Sampling params: {sampling_params}") return sampling_params @@ -534,7 +552,9 @@ async def _generate_disaggregated( if not input_ids: raise ValueError("input_ids is required") - sampling_params = SglangUtils.build_sampling_params(request) + sampling_params = SglangUtils.build_sampling_params( + request, use_sglang_tokenizer=self.use_sglang_tokenizer + ) # Request bootstrap info from prefill worker bootstrap_info = await self._get_bootstrap_from_prefill( @@ -583,7 +603,9 @@ async def _generate_aggregated( raise ValueError("input_ids is required") tensor_id: int | None = None try: - sampling_params = SglangUtils.build_sampling_params(request) + sampling_params = SglangUtils.build_sampling_params( + request, use_sglang_tokenizer=self.use_sglang_tokenizer + ) with _nvtx.annotate("mm:pd:load_multimodal", color="cyan"): ( image_mm_items, diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index 0aa051b91c1a..4e7e96ff9795 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -799,7 +799,7 @@ 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(): +def test_build_sampling_params_enforces_min_tokens_in_dynamo_for_token_requests(): handler = _new_decode_handler(use_sglang_tokenizer=False) sampling_params = handler._build_sampling_params( @@ -808,14 +808,19 @@ def test_build_sampling_params_maps_min_tokens_for_token_requests(): "stop_conditions": { "max_tokens": 64, "min_tokens": 64, - "ignore_eos": True, + "stop_token_ids": [7], + "stop_token_ids_hidden": [151643], }, } ) - assert sampling_params["min_new_tokens"] == 64 - assert sampling_params["max_new_tokens"] == 64 + # SGLang rejects min_new_tokens when the engine has no tokenizer, and Dynamo's + # decoder already enforces the floor, so the engine is held open instead of told + # about it: no stop ids, ignore_eos set, over-generation bounded by max_tokens. + 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 def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests(): @@ -839,32 +844,56 @@ def test_build_sampling_params_omits_min_new_tokens_when_min_tokens_absent(): assert "min_new_tokens" not in sampling_params -def test_build_sampling_params_forwards_explicit_zero_min_tokens(): +def test_build_sampling_params_leaves_engine_stopping_alone_for_zero_min_tokens(): handler = _new_decode_handler(use_sglang_tokenizer=False) sampling_params = handler._build_sampling_params( - {"sampling_options": {}, "stop_conditions": {"max_tokens": 8, "min_tokens": 0}} + { + "sampling_options": {}, + "stop_conditions": { + "max_tokens": 8, + "min_tokens": 0, + "stop_token_ids": [7], + }, + } ) - assert sampling_params["min_new_tokens"] == 0 + # A zero floor asks for nothing, so the engine keeps its own stop conditions. + assert "min_new_tokens" not in sampling_params + assert sampling_params["stop_token_ids"] == [7] + assert "ignore_eos" not in sampling_params -def test_multimodal_build_sampling_params_maps_min_tokens(): - request = SglangMultimodalRequest( +def _multimodal_min_tokens_request() -> SglangMultimodalRequest: + return SglangMultimodalRequest( request=PreprocessedRequest( token_ids=[1, 2, 3], - stop_conditions=StopConditions( - max_tokens=64, min_tokens=64, ignore_eos=True - ), + stop_conditions=StopConditions(max_tokens=64, min_tokens=64), sampling_options=SamplingOptions(), ) ) - sampling_params = SglangUtils.build_sampling_params(request) + +def test_multimodal_build_sampling_params_holds_engine_open_for_min_tokens(): + sampling_params = SglangUtils.build_sampling_params( + _multimodal_min_tokens_request(), use_sglang_tokenizer=False + ) + + # Tokenizer-free engine: Dynamo's decoder enforces the floor, so the engine is only + # kept from stopping early. Over-generation is bounded by max_tokens. + assert "min_new_tokens" not in sampling_params + assert sampling_params["ignore_eos"] is True + assert sampling_params["max_new_tokens"] == 64 + + +def test_multimodal_build_sampling_params_maps_min_tokens_with_engine_tokenizer(): + sampling_params = SglangUtils.build_sampling_params( + _multimodal_min_tokens_request(), use_sglang_tokenizer=True + ) assert sampling_params["min_new_tokens"] == 64 assert sampling_params["max_new_tokens"] == 64 - assert sampling_params["ignore_eos"] is True + assert "ignore_eos" not in sampling_params @pytest.mark.parametrize( diff --git a/lib/sidecar/sglang/src/protocol.rs b/lib/sidecar/sglang/src/protocol.rs index b9afe744c5fb..31405bf60857 100644 --- a/lib/sidecar/sglang/src/protocol.rs +++ b/lib/sidecar/sglang/src/protocol.rs @@ -40,16 +40,16 @@ pub(crate) fn build_generate_request( .transpose() .map_err(|_| client::invalid_arg("max_tokens does not fit in i32"))? }; - let min_new_tokens = if mode.is_prefill() { - None - } else { - request - .stop_conditions - .min_tokens - .map(i32::try_from) - .transpose() - .map_err(|_| client::invalid_arg("min_tokens does not fit in i32"))? - }; + // `min_new_tokens` is never sent: SGLang rejects it outright when the engine has no + // tokenizer, and this gRPC path always submits token ids, so it never has one. Dynamo + // enforces the floor itself -- the `Decoder` in `lib/llm/src/backend.rs` holds the same + // `min_tokens` and the same stop-token set, and suppresses every stop trigger until the + // floor is reached. What the `Decoder` cannot do is manufacture tokens the engine never + // produced, so when a floor is requested the engine is held open for the whole request: + // it receives no stop ids and is told to ignore EOS, and termination becomes entirely + // Dynamo's. The engine may then generate tokens Dynamo discards, bounded by `max_tokens`. + let hold_engine_open = + !mode.is_prefill() && request.stop_conditions.min_tokens.unwrap_or(0) > 0; let mut stop_token_ids = Vec::new(); for tokens in [ @@ -68,6 +68,9 @@ pub(crate) fn build_generate_request( } } } + if hold_engine_open { + stop_token_ids.clear(); + } let guided = request.sampling_options.guided_decoding.as_ref(); let sampling_params = pb::SamplingParams { @@ -79,10 +82,14 @@ pub(crate) fn build_generate_request( presence_penalty: request.sampling_options.presence_penalty, repetition_penalty: request.sampling_options.repetition_penalty, max_new_tokens, - min_new_tokens, + min_new_tokens: None, stop: request.stop_conditions.stop.clone().unwrap_or_default(), stop_token_ids, - ignore_eos: request.stop_conditions.ignore_eos, + ignore_eos: if hold_engine_open { + Some(true) + } else { + request.stop_conditions.ignore_eos + }, n: request.sampling_options.n.map(i32::from), json_schema: guided .and_then(|value| value.json.as_ref()) @@ -649,6 +656,55 @@ mod tests { assert_eq!(mapped.disaggregated_params.unwrap().bootstrap_port, 5001); } + #[test] + fn decode_min_tokens_is_left_to_the_frontend_and_engine_is_held_open() { + let mut request = request(); + request.stop_conditions.min_tokens = Some(4); + request.stop_conditions.stop_token_ids = Some(vec![7]); + request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); + + let mapped = build_generate_request( + &request, + "rid-min", + DisaggregationMode::Aggregated, + None, + None, + ) + .unwrap(); + let sampling = mapped.sampling_params.unwrap(); + + // SGLang refuses min_new_tokens when it has no tokenizer, and it never has one on + // this path, so the parameter is not sent. The Decoder holds the same min_tokens. + assert_eq!(sampling.min_new_tokens, None); + // The Decoder can only withhold stops, not manufacture tokens, so the engine must be + // held open past the threshold: it gets no stop ids and is told to ignore EOS. + assert!(sampling.stop_token_ids.is_empty()); + assert_eq!(sampling.ignore_eos, Some(true)); + // max_tokens still bounds the over-generation this buys. + assert_eq!(sampling.max_new_tokens, Some(8)); + } + + #[test] + fn decode_without_min_tokens_keeps_engine_side_stopping() { + let mut request = request(); + request.stop_conditions.stop_token_ids = Some(vec![7]); + request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); + + let mapped = build_generate_request( + &request, + "rid-no-min", + DisaggregationMode::Aggregated, + None, + None, + ) + .unwrap(); + let sampling = mapped.sampling_params.unwrap(); + + assert_eq!(sampling.min_new_tokens, None); + assert_eq!(sampling.stop_token_ids, vec![7, 151643]); + assert_eq!(sampling.ignore_eos, None); + } + #[test] fn prefill_uses_selected_prefill_dp_rank() { let mut request = request(); From 651ba2c9fbf79e2648f5b0096840a29db22c5938 Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 3 Sep 2026 18:42:07 +0000 Subject: [PATCH 02/15] chore: remove factory working notes Signed-off-by: svc-glamr@nvidia.com --- .../sglang/request_handlers/llm/decode_handler.py | 10 ++-------- .../request_handlers/multimodal/worker_handler.py | 8 ++------ .../dynamo/sglang/tests/test_sglang_decode_handler.py | 5 ++--- lib/sidecar/sglang/src/protocol.rs | 10 ++-------- 4 files changed, 8 insertions(+), 25 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index 69acc82d75d8..082002649a79 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -434,14 +434,8 @@ 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 - # min_new_tokens is never sent on this branch. SGLang rejects it when the - # engine has no tokenizer, which is exactly what this branch means, and - # Dynamo already enforces the floor: the Rust decoder holds the same - # min_tokens and the same stop-token set, and suppresses every stop until - # the floor is reached. It cannot manufacture tokens the engine never - # produced, so when a floor is requested the engine is held open instead — - # no stop ids, ignore_eos set — and termination becomes entirely Dynamo's. - # The engine may then generate tokens Dynamo discards, bounded by max_tokens. + # SGLang rejects min_new_tokens without a tokenizer, so Dynamo's decoder + # enforces the floor and the engine is held open until it stops the request. hold_engine_open = (stop_conditions.get("min_tokens") or 0) > 0 if hold_engine_open: stop_token_ids = None diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index c73d28825264..9e1e98f92b83 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -101,12 +101,8 @@ def build_sampling_params( sampling_params["n"] = sampling_options.n if stop_conditions.max_tokens: sampling_params["max_new_tokens"] = stop_conditions.max_tokens - # min_new_tokens only goes to an engine that owns a tokenizer; SGLang rejects it - # otherwise. Dynamo already enforces the floor -- the Rust decoder holds the same - # min_tokens and the same stop-token set and suppresses every stop until the floor - # is reached -- but it cannot manufacture tokens the engine never produced, so a - # tokenizer-free engine is held open with ignore_eos and left to Dynamo to stop. - # The engine may then generate tokens Dynamo discards, bounded by max_tokens. + # SGLang rejects min_new_tokens without a tokenizer, so Dynamo's decoder enforces + # the floor and the engine is held open with ignore_eos until it stops the request. hold_engine_open = False if stop_conditions.min_tokens: if use_sglang_tokenizer: diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index 4e7e96ff9795..8c1ee91ad8b2 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -814,9 +814,8 @@ def test_build_sampling_params_enforces_min_tokens_in_dynamo_for_token_requests( } ) - # SGLang rejects min_new_tokens when the engine has no tokenizer, and Dynamo's - # decoder already enforces the floor, so the engine is held open instead of told - # about it: no stop ids, ignore_eos set, over-generation bounded by max_tokens. + # Tokenizer-free: Dynamo's decoder enforces the floor, so the engine is only held + # open -- no min_new_tokens, no stop ids, ignore_eos set. assert "min_new_tokens" not in sampling_params assert "stop_token_ids" not in sampling_params assert sampling_params["ignore_eos"] is True diff --git a/lib/sidecar/sglang/src/protocol.rs b/lib/sidecar/sglang/src/protocol.rs index 31405bf60857..49ddac7f66e8 100644 --- a/lib/sidecar/sglang/src/protocol.rs +++ b/lib/sidecar/sglang/src/protocol.rs @@ -40,14 +40,8 @@ pub(crate) fn build_generate_request( .transpose() .map_err(|_| client::invalid_arg("max_tokens does not fit in i32"))? }; - // `min_new_tokens` is never sent: SGLang rejects it outright when the engine has no - // tokenizer, and this gRPC path always submits token ids, so it never has one. Dynamo - // enforces the floor itself -- the `Decoder` in `lib/llm/src/backend.rs` holds the same - // `min_tokens` and the same stop-token set, and suppresses every stop trigger until the - // floor is reached. What the `Decoder` cannot do is manufacture tokens the engine never - // produced, so when a floor is requested the engine is held open for the whole request: - // it receives no stop ids and is told to ignore EOS, and termination becomes entirely - // Dynamo's. The engine may then generate tokens Dynamo discards, bounded by `max_tokens`. + // SGLang rejects `min_new_tokens` without a tokenizer and this path never has one, so the + // `Decoder` in `lib/llm/src/backend.rs` enforces the floor and the engine is held open. let hold_engine_open = !mode.is_prefill() && request.stop_conditions.min_tokens.unwrap_or(0) > 0; From 9b469d7fa4e6fb07d20cc75ad09adaee7489a7eb Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 3 Sep 2026 19:22:58 +0000 Subject: [PATCH 03/15] fix(sglang): clear stop strings on the held-open sidecar path When a tokenizer-free decode request carries a min_tokens floor, the sidecar holds the engine open so Dynamo's Decoder can enforce the floor. Stop token ids were cleared for that, but stop strings were still forwarded, so a matching stop string could end generation below the floor. Clear both in the same branch. Also make the multimodal handler use the same positive min_tokens predicate as the decode handler and the sidecar. Signed-off-by: svc-glamr@nvidia.com --- .../request_handlers/multimodal/worker_handler.py | 2 +- .../sglang/tests/test_sglang_decode_handler.py | 1 - lib/sidecar/sglang/src/protocol.rs | 13 +++++++++++-- 3 files changed, 12 insertions(+), 4 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index 9e1e98f92b83..9b6c3f7f797b 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -104,7 +104,7 @@ def build_sampling_params( # SGLang rejects min_new_tokens without a tokenizer, so Dynamo's decoder enforces # the floor and the engine is held open with ignore_eos until it stops the request. hold_engine_open = False - if stop_conditions.min_tokens: + if (stop_conditions.min_tokens or 0) > 0: if use_sglang_tokenizer: sampling_params["min_new_tokens"] = stop_conditions.min_tokens else: diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index 8c1ee91ad8b2..938ae139fd3d 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -857,7 +857,6 @@ def test_build_sampling_params_leaves_engine_stopping_alone_for_zero_min_tokens( } ) - # A zero floor asks for nothing, so the engine keeps its own stop conditions. assert "min_new_tokens" not in sampling_params assert sampling_params["stop_token_ids"] == [7] assert "ignore_eos" not in sampling_params diff --git a/lib/sidecar/sglang/src/protocol.rs b/lib/sidecar/sglang/src/protocol.rs index 49ddac7f66e8..1b4f4eccf216 100644 --- a/lib/sidecar/sglang/src/protocol.rs +++ b/lib/sidecar/sglang/src/protocol.rs @@ -62,8 +62,12 @@ pub(crate) fn build_generate_request( } } } + let mut stop = request.stop_conditions.stop.clone().unwrap_or_default(); if hold_engine_open { + // The `Decoder` withholds every stop until the floor is reached, stop strings + // included, so any stop left with the engine can end the stream below the floor. stop_token_ids.clear(); + stop.clear(); } let guided = request.sampling_options.guided_decoding.as_ref(); @@ -77,7 +81,7 @@ pub(crate) fn build_generate_request( repetition_penalty: request.sampling_options.repetition_penalty, max_new_tokens, min_new_tokens: None, - stop: request.stop_conditions.stop.clone().unwrap_or_default(), + stop, stop_token_ids, ignore_eos: if hold_engine_open { Some(true) @@ -656,6 +660,7 @@ mod tests { request.stop_conditions.min_tokens = Some(4); request.stop_conditions.stop_token_ids = Some(vec![7]); request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); + request.stop_conditions.stop = Some(vec!["STOP".to_string()]); let mapped = build_generate_request( &request, @@ -671,8 +676,10 @@ mod tests { // this path, so the parameter is not sent. The Decoder holds the same min_tokens. assert_eq!(sampling.min_new_tokens, None); // The Decoder can only withhold stops, not manufacture tokens, so the engine must be - // held open past the threshold: it gets no stop ids and is told to ignore EOS. + // held open past the threshold: it gets no stop ids, no stop strings, and is told to + // ignore EOS. assert!(sampling.stop_token_ids.is_empty()); + assert!(sampling.stop.is_empty()); assert_eq!(sampling.ignore_eos, Some(true)); // max_tokens still bounds the over-generation this buys. assert_eq!(sampling.max_new_tokens, Some(8)); @@ -683,6 +690,7 @@ mod tests { let mut request = request(); request.stop_conditions.stop_token_ids = Some(vec![7]); request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); + request.stop_conditions.stop = Some(vec!["STOP".to_string()]); let mapped = build_generate_request( &request, @@ -696,6 +704,7 @@ mod tests { assert_eq!(sampling.min_new_tokens, None); assert_eq!(sampling.stop_token_ids, vec![7, 151643]); + assert_eq!(sampling.stop, vec!["STOP".to_string()]); assert_eq!(sampling.ignore_eos, None); } From a768779c11c8e3eb0042ef75a885f641392501fc Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 3 Sep 2026 19:37:17 +0000 Subject: [PATCH 04/15] test(sglang): cover a floor below the ceiling and the stop after it The tokenizer-free lowering tests all used min_tokens equal to max_tokens, which cannot tell a Dynamo-enforced floor apart from the engine stopping at its own maximum. Add a case with the floor below the ceiling and a case with no ceiling at all, where max_new_tokens stays None on purpose. Pair min_tokens_delays_stop with the other half of the contract: once the floor is reached, the decoder does stop at EOS. Signed-off-by: svc-glamr@nvidia.com --- .../tests/test_sglang_decode_handler.py | 46 +++++++++++++++++++ lib/llm/tests/test_stop_behavior.rs | 12 +++++ 2 files changed, 58 insertions(+) diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index 938ae139fd3d..b71fd0381621 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -822,6 +822,52 @@ def test_build_sampling_params_enforces_min_tokens_in_dynamo_for_token_requests( assert sampling_params["max_new_tokens"] == 64 +def test_build_sampling_params_holds_engine_open_below_max_tokens(): + handler = _new_decode_handler(use_sglang_tokenizer=False) + + sampling_params = handler._build_sampling_params( + { + "sampling_options": {}, + "stop_conditions": { + "max_tokens": 64, + "min_tokens": 8, + "stop_token_ids": [7], + "stop_token_ids_hidden": [151643], + }, + } + ) + + # A floor below the ceiling is the case that separates the two: the engine still runs + # to max_new_tokens, and only Dynamo's decoder knows where the floor is. + 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 + + +def test_build_sampling_params_holds_engine_open_without_max_tokens(): + handler = _new_decode_handler(use_sglang_tokenizer=False) + + sampling_params = handler._build_sampling_params( + { + "sampling_options": {}, + "stop_conditions": { + "min_tokens": 8, + "stop_token_ids_hidden": [151643], + }, + } + ) + + # max_new_tokens is kept as None deliberately -- SGLang reads that as "until EOS or + # context length", and omitting it would instead impose a default of 128, which is + # below the floor a caller may have asked for. With no ceiling and ignore_eos set, + # context length is what bounds the over-generation this buys. + assert sampling_params["max_new_tokens"] is None + assert "min_new_tokens" not in sampling_params + assert "stop_token_ids" not in sampling_params + assert sampling_params["ignore_eos"] is True + + def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests(): handler = _new_decode_handler(use_sglang_tokenizer=True) diff --git a/lib/llm/tests/test_stop_behavior.rs b/lib/llm/tests/test_stop_behavior.rs index e61e2612eec3..205fcc29c953 100644 --- a/lib/llm/tests/test_stop_behavior.rs +++ b/lib/llm/tests/test_stop_behavior.rs @@ -125,6 +125,18 @@ fn min_tokens_delays_stop() { assert!(result.stop_trigger.is_none()); } +#[test] +fn min_tokens_stops_at_eos_once_floor_is_reached() { + let mut decoder = make_decoder(None, Some(2), Some(vec![EOS]), None, false); + let result = decoder.process_token_ids(&[HI, THERE, EOS]).unwrap(); + + assert_eq!(result.text.as_deref(), Some("hithere")); + assert!(matches!( + result.stop_trigger, + Some(StopTrigger::HiddenStopTokenDetected(id)) if id == EOS + )); +} + #[test] fn stop_token_priority_over_sequence() { let mut decoder = make_decoder(None, None, Some(vec![STOP]), Some(vec!["STOP"]), false); From e815d202d06cdaba6e8be060bd848075cd2e22ae Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Thu, 3 Sep 2026 20:09:08 +0000 Subject: [PATCH 05/15] style(sglang): drop narrating comments from the sidecar lowering tests The surrounding tests in this module assert without commentary, and these lines restated both the assertion below them and the rationale already recorded next to the implementation. Signed-off-by: svc-glamr@nvidia.com --- lib/sidecar/sglang/src/protocol.rs | 6 ------ 1 file changed, 6 deletions(-) diff --git a/lib/sidecar/sglang/src/protocol.rs b/lib/sidecar/sglang/src/protocol.rs index 1b4f4eccf216..db6ea9bf697b 100644 --- a/lib/sidecar/sglang/src/protocol.rs +++ b/lib/sidecar/sglang/src/protocol.rs @@ -672,16 +672,10 @@ mod tests { .unwrap(); let sampling = mapped.sampling_params.unwrap(); - // SGLang refuses min_new_tokens when it has no tokenizer, and it never has one on - // this path, so the parameter is not sent. The Decoder holds the same min_tokens. assert_eq!(sampling.min_new_tokens, None); - // The Decoder can only withhold stops, not manufacture tokens, so the engine must be - // held open past the threshold: it gets no stop ids, no stop strings, and is told to - // ignore EOS. assert!(sampling.stop_token_ids.is_empty()); assert!(sampling.stop.is_empty()); assert_eq!(sampling.ignore_eos, Some(true)); - // max_tokens still bounds the over-generation this buys. assert_eq!(sampling.max_new_tokens, Some(8)); } From 3427ee062639bd5fe7f9f560b179c9d54179ba1e Mon Sep 17 00:00:00 2001 From: GLAMR Date: Fri, 11 Sep 2026 22:07:13 +0000 Subject: [PATCH 06/15] fix(sglang): route native min-token requests through decoder Signed-off-by: GLAMR --- lib/sidecar/sglang/src/native_http.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/lib/sidecar/sglang/src/native_http.rs b/lib/sidecar/sglang/src/native_http.rs index 5c1000fb9f63..f4e3127ed394 100644 --- a/lib/sidecar/sglang/src/native_http.rs +++ b/lib/sidecar/sglang/src/native_http.rs @@ -47,6 +47,11 @@ pub(crate) fn request( else { return Ok(None); }; + if !mode.is_prefill() && request.stop_conditions.min_tokens.unwrap_or(0) > 0 { + // Native HTTP responses bypass Dynamo's Decoder, which owns minimum-token + // enforcement when SGLang runs without a tokenizer. + return Ok(None); + } let mut body = payload .as_object() .cloned() From 32694da21df5de93699163896f16dd469bfbacbf Mon Sep 17 00:00:00 2001 From: "svc-glamr@nvidia.com" Date: Fri, 11 Sep 2026 23:01:37 +0000 Subject: [PATCH 07/15] test(llm): pass the jail seed argument to Decoder::new in the floor test Decoder::new takes a fifth jail_seed parameter. The visible-stop floor test still called it with four arguments, so the dynamo-llm test target failed to compile. Signed-off-by: svc-glamr@nvidia.com --- lib/llm/tests/test_stop_behavior.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/lib/llm/tests/test_stop_behavior.rs b/lib/llm/tests/test_stop_behavior.rs index d20e868fac48..0f2b7e9d7c5c 100644 --- a/lib/llm/tests/test_stop_behavior.rs +++ b/lib/llm/tests/test_stop_behavior.rs @@ -203,7 +203,7 @@ fn min_tokens_keeps_visible_stop_tokens_below_the_floor() { stop_token_ids_visible: Some(vec![STOP]), ..Default::default() }; - let mut decoder = Decoder::new(decode_stream, stop_conditions, false, None); + let mut decoder = Decoder::new(decode_stream, stop_conditions, false, None, None); let result = decoder.process_token_ids(&[HI, STOP]).unwrap(); assert_eq!(result.text.as_deref(), Some("hiSTOP")); From b482082beb8578d78ed885b4a7adb740e561d691 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Sat, 12 Sep 2026 00:34:05 +0000 Subject: [PATCH 08/15] fix(sglang): abort multimodal requests before first chunk Signed-off-by: GLAMR --- .../sglang/request_handlers/multimodal/worker_handler.py | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index 16bec1fa091b..2b45b78dd627 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -599,6 +599,8 @@ async def _generate_disaggregated( rng_first = _nvtx.start_range("mm:dec:first_token", color="purple") first_token = True request_id_future: "asyncio.Future[str]" = asyncio.Future() + if context is not None and not request_id_future.done(): + request_id_future.set_result(context.trace_id) try: async with self._engine_abort_on_cancel(request_id_future, context): async for output in StreamProcessor.process_sglang_stream( @@ -669,6 +671,8 @@ async def _generate_aggregated( rng_first = _nvtx.start_range("mm:dec:first_token", color="purple") first_token = True request_id_future: "asyncio.Future[str]" = asyncio.Future() + if context is not None and not request_id_future.done(): + request_id_future.set_result(context.trace_id) try: async with self._engine_abort_on_cancel(request_id_future, context): async for output in StreamProcessor.process_sglang_stream( From 2492a7693e68193af895b4f83a77ed1af3e21ca6 Mon Sep 17 00:00:00 2001 From: jain-ria Date: Tue, 15 Sep 2026 13:30:21 -0700 Subject: [PATCH 09/15] fix(sglang): narrow tokenizer-free min token fallback Signed-off-by: jain-ria --- .../sglang/request_handlers/handler_base.py | 24 -- .../request_handlers/llm/decode_handler.py | 17 +- .../multimodal/worker_handler.py | 106 +++------ .../tests/test_sglang_decode_handler.py | 159 +++----------- .../tests/test_sglang_multimodal_video.py | 1 - lib/llm/src/backend.rs | 205 ++++++------------ lib/llm/tests/test_stop_behavior.rs | 70 +----- lib/sidecar/sglang/src/native_http.rs | 5 - lib/sidecar/sglang/src/protocol.rs | 79 ++----- 9 files changed, 150 insertions(+), 516 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index 758b020a479c..26cf4e1fd0e8 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -589,14 +589,6 @@ def __init__( self.kv_publisher = publisher.kv_publisher self.serving_mode = config.serving_mode self.use_sglang_tokenizer = config.dynamo_args.use_sglang_tokenizer - # Whether the engine itself was started without a tokenizer. This is a - # different fact from use_sglang_tokenizer, which only records who owns - # preprocessing: --dyn-chat-processor sglang leaves preprocessing to - # Dynamo while the engine keeps its tokenizer. Read defensively — the - # diffusion workers pass a SimpleNamespace stub for server_args. - self.engine_skips_tokenizer = bool( - getattr(config.server_args, "skip_tokenizer_init", False) - ) self.enable_trace = getattr(config.server_args, "enable_trace", False) self._max_input_token_id: Optional[int] = None @@ -1339,22 +1331,6 @@ async def _handle_cancellation( except (asyncio.CancelledError, Exception): pass - @asynccontextmanager - async def _engine_abort_on_cancel( - self, request_id_future: asyncio.Future, context: Optional[Context] - ) -> AsyncGenerator[None, None]: - """Abort the engine request once the caller stops consuming the stream. - - This is ``_cancellation_monitor`` for callers that want the abort but - have no use for the monitoring task, and that may be handed no context - at all — in which case there is nothing to watch and nothing to abort. - """ - if context is None: - yield - return - async with self._cancellation_monitor(request_id_future, context): - yield - @asynccontextmanager async def _cancellation_monitor( self, request_id_future: asyncio.Future, context: Context diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index b86608f9d83e..ab184517070c 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -361,19 +361,22 @@ 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 min_new_tokens, so Dynamo's decoder - # enforces the floor and the engine is held open until it stops the - # request. An engine that kept its tokenizer takes min_new_tokens - # directly and keeps its own stopping. - min_tokens = stop_conditions.get("min_tokens") or 0 - hold_engine_open = min_tokens > 0 and self.engine_skips_tokenizer + # 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": None if hold_engine_open else min_tokens or None, + "min_new_tokens": None if hold_engine_open else min_tokens, "ignore_eos": True if hold_engine_open else stop_conditions.get("ignore_eos"), diff --git a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py index 2b45b78dd627..3b60546481da 100644 --- a/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py +++ b/components/src/dynamo/sglang/request_handlers/multimodal/worker_handler.py @@ -77,17 +77,8 @@ class SglangUtils: """General SGLang utilities (not multimodal-specific)""" @staticmethod - def build_sampling_params( - request: SglangMultimodalRequest, *, engine_skips_tokenizer: bool - ) -> dict: - """Build sampling parameters for SGLang engine (generic functionality) - - Args: - request: Pre-tokenized multimodal request. - engine_skips_tokenizer: Whether the engine was started without a - tokenizer. When it was, min_tokens is enforced by Dynamo rather - than forwarded to the engine. - """ + def build_sampling_params(request: SglangMultimodalRequest) -> dict: + """Build sampling parameters for SGLang engine (generic functionality)""" sampling_params = {} # Extract sampling options from request @@ -102,21 +93,12 @@ def build_sampling_params( sampling_params["top_k"] = sampling_options.top_k if sampling_options.n is not None: sampling_params["n"] = sampling_options.n - # Always set max_new_tokens, including when it is None: SGLang reads None - # as "generate until EOS or the context limit", whereas omitting the key - # falls back to a default of 128 and would cap a request that asked only - # for a floor. - sampling_params["max_new_tokens"] = stop_conditions.max_tokens - # A tokenizer-free SGLang rejects min_new_tokens, so Dynamo's decoder enforces - # the floor and the engine is held open with ignore_eos until it stops the request. - hold_engine_open = False - if (stop_conditions.min_tokens or 0) > 0: - if engine_skips_tokenizer: - hold_engine_open = True - else: - sampling_params["min_new_tokens"] = stop_conditions.min_tokens - if hold_engine_open or stop_conditions.ignore_eos: - sampling_params["ignore_eos"] = True + if stop_conditions.max_tokens: + sampling_params["max_new_tokens"] = stop_conditions.max_tokens + if stop_conditions.min_tokens: + sampling_params["min_new_tokens"] = stop_conditions.min_tokens + if stop_conditions.ignore_eos: + sampling_params["ignore_eos"] = stop_conditions.ignore_eos logger.debug(f"Sampling params: {sampling_params}") return sampling_params @@ -225,32 +207,18 @@ class StreamProcessor: """Unified stream processing for SGLang responses""" @staticmethod - async def process_sglang_stream( - stream_source, - request_id_future: Optional["asyncio.Future[str]"] = None, - ) -> AsyncIterator[str]: + async def process_sglang_stream(stream_source) -> AsyncIterator[str]: """Process SGLang stream output. With stream_output=True (enforced by Dynamo), SGLang sends disjoint segments containing only new tokens since the last output. We pass these through directly. - - Args: - stream_source: Async iterator of SGLang response dicts. - request_id_future: Resolved with the SGLang request id from the first - chunk that carries one, so a cancellation monitor can abort that - specific request. """ try: async for res in stream_source: try: - meta_info = res.get("meta_info", {}) - if request_id_future is not None and not request_id_future.done(): - sglang_request_id = meta_info.get("id") - if sglang_request_id: - request_id_future.set_result(sglang_request_id) # With stream_output=True, output_ids contains only new tokens (disjoint) output_ids = res.get("output_ids", []) - finish_reason = meta_info.get("finish_reason") + finish_reason = res.get("meta_info", {}).get("finish_reason") # Empty, non-final chunks can happen during scheduler idle ticks. # Keep waiting for the next chunk. @@ -570,9 +538,7 @@ async def _generate_disaggregated( if not input_ids: raise ValueError("input_ids is required") - sampling_params = SglangUtils.build_sampling_params( - request, engine_skips_tokenizer=self.engine_skips_tokenizer - ) + sampling_params = SglangUtils.build_sampling_params(request) validate_disagg_parallel_sampling({"sampling_params": sampling_params}) # Request bootstrap info from prefill worker @@ -598,21 +564,13 @@ async def _generate_disaggregated( rng_first = _nvtx.start_range("mm:dec:first_token", color="purple") first_token = True - request_id_future: "asyncio.Future[str]" = asyncio.Future() - if context is not None and not request_id_future.done(): - request_id_future.set_result(context.trace_id) try: - async with self._engine_abort_on_cancel(request_id_future, context): - async for output in StreamProcessor.process_sglang_stream( - decode_stream, request_id_future=request_id_future - ): - if first_token: - end_ttft() - _nvtx.end_range(rng_first) - first_token = False - yield output - if context is not None and context.is_stopped(): - break + async for output in StreamProcessor.process_sglang_stream(decode_stream): + if first_token: + end_ttft() + _nvtx.end_range(rng_first) + first_token = False + yield output finally: if first_token: end_ttft() @@ -630,9 +588,7 @@ async def _generate_aggregated( raise ValueError("input_ids is required") tensor_id: int | None = None try: - sampling_params = SglangUtils.build_sampling_params( - request, engine_skips_tokenizer=self.engine_skips_tokenizer - ) + sampling_params = SglangUtils.build_sampling_params(request) with _nvtx.annotate("mm:pd:load_multimodal", color="cyan"): ( image_mm_items, @@ -670,24 +626,16 @@ async def _generate_aggregated( rng_first = _nvtx.start_range("mm:dec:first_token", color="purple") first_token = True - request_id_future: "asyncio.Future[str]" = asyncio.Future() - if context is not None and not request_id_future.done(): - request_id_future.set_result(context.trace_id) try: - async with self._engine_abort_on_cancel(request_id_future, context): - async for output in StreamProcessor.process_sglang_stream( - agg_stream, request_id_future=request_id_future - ): - if first_token: - if tensor_id is not None: - self.embeddings_processor.release_embeddings(tensor_id) - tensor_id = None - end_ttft() - _nvtx.end_range(rng_first) - first_token = False - yield output - if context is not None and context.is_stopped(): - break + async for output in StreamProcessor.process_sglang_stream(agg_stream): + if first_token: + if tensor_id is not None: + self.embeddings_processor.release_embeddings(tensor_id) + tensor_id = None + end_ttft() + _nvtx.end_range(rng_first) + first_token = False + yield output finally: if first_token: end_ttft() diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index fde9f162bb20..80619b57b19d 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -245,22 +245,16 @@ def test_openai_stop_sampling_params_maps_token_id_stop_array(): def _new_decode_handler( *, use_sglang_tokenizer: bool = False, - engine_skips_tokenizer: bool | None = None, + skip_tokenizer_init: bool = False, enable_rl: bool = False, ): - # The two are separate facts. Default to the common pairing -- Dynamo owning - # preprocessing for an engine started with --skip-tokenizer-init -- and let a - # test name engine_skips_tokenizer when it wants them to disagree. - if engine_skips_tokenizer is None: - engine_skips_tokenizer = not use_sglang_tokenizer handler = DecodeWorkerHandler.__new__(DecodeWorkerHandler) handler.shutdown_event = None handler.use_sglang_tokenizer = use_sglang_tokenizer - handler.engine_skips_tokenizer = engine_skips_tokenizer handler.config = SimpleNamespace( server_args=SimpleNamespace( served_model_name="test-model", - skip_tokenizer_init=engine_skips_tokenizer, + skip_tokenizer_init=skip_tokenizer_init, ), dynamo_args=SimpleNamespace(enable_rl=enable_rl), ) @@ -827,32 +821,12 @@ def test_guided_decoding_params_are_accepted_by_sglang(guided_decoding): SamplingParams(**sampling_params) -def test_build_sampling_params_enforces_min_tokens_in_dynamo_for_token_requests(): - handler = _new_decode_handler(use_sglang_tokenizer=False) - - sampling_params = handler._build_sampling_params( - { - "sampling_options": {}, - "stop_conditions": { - "max_tokens": 64, - "min_tokens": 64, - "stop_token_ids": [7], - "stop_token_ids_hidden": [151643], - }, - } +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, ) - # Tokenizer-free: Dynamo's decoder enforces the floor, so the engine is only held - # open -- no min_new_tokens, no stop ids, ignore_eos set. - 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 - - -def test_build_sampling_params_holds_engine_open_below_max_tokens(): - handler = _new_decode_handler(use_sglang_tokenizer=False) - sampling_params = handler._build_sampling_params( { "sampling_options": {}, @@ -865,53 +839,20 @@ def test_build_sampling_params_holds_engine_open_below_max_tokens(): } ) - # A floor below the ceiling is the case that separates the two: the engine still runs - # to max_new_tokens, and only Dynamo's decoder knows where the floor is. 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 -def test_build_sampling_params_holds_engine_open_without_max_tokens(): - handler = _new_decode_handler(use_sglang_tokenizer=False) - - sampling_params = handler._build_sampling_params( - { - "sampling_options": {}, - "stop_conditions": { - "min_tokens": 8, - "stop_token_ids_hidden": [151643], - }, - } - ) - - # max_new_tokens is kept as None deliberately -- SGLang reads that as "until EOS or - # context length", and omitting it would instead impose a default of 128, which is - # below the floor a caller may have asked for. With no ceiling and ignore_eos set, - # context length is what bounds the over-generation this buys. - assert sampling_params["max_new_tokens"] is None - assert "min_new_tokens" not in sampling_params - assert "stop_token_ids" not in sampling_params - assert sampling_params["ignore_eos"] is True - - -def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests(): - handler = _new_decode_handler(use_sglang_tokenizer=True) + SamplingParams(**sampling_params).normalize(tokenizer=None) - sampling_params = handler._build_sampling_params( - {"max_tokens": 64, "min_tokens": 16} - ) - - assert sampling_params["min_new_tokens"] == 16 - assert sampling_params["max_new_tokens"] == 64 - -def test_build_sampling_params_forwards_min_tokens_to_an_engine_with_a_tokenizer(): - # Dynamo owns preprocessing, but the engine was started with its tokenizer, so it - # accepts min_new_tokens and there is no reason to take its stopping away. +def test_build_sampling_params_forwards_min_tokens_when_engine_has_tokenizer(): handler = _new_decode_handler( - use_sglang_tokenizer=False, engine_skips_tokenizer=False + use_sglang_tokenizer=False, + skip_tokenizer_init=False, ) sampling_params = handler._build_sampling_params( @@ -931,86 +872,58 @@ def test_build_sampling_params_forwards_min_tokens_to_an_engine_with_a_tokenizer assert sampling_params["max_new_tokens"] == 64 -def test_build_sampling_params_omits_min_new_tokens_when_min_tokens_absent(): - handler = _new_decode_handler(use_sglang_tokenizer=False) +def test_build_sampling_params_maps_min_tokens_for_sglang_tokenizer_requests(): + handler = _new_decode_handler(use_sglang_tokenizer=True) sampling_params = handler._build_sampling_params( - {"sampling_options": {}, "stop_conditions": {"max_tokens": 8}} + {"max_tokens": 64, "min_tokens": 16} ) - assert "min_new_tokens" not in sampling_params + assert sampling_params["min_new_tokens"] == 16 + assert sampling_params["max_new_tokens"] == 64 -def test_build_sampling_params_leaves_engine_stopping_alone_for_zero_min_tokens(): - handler = _new_decode_handler(use_sglang_tokenizer=False) +def test_build_sampling_params_omits_min_new_tokens_when_min_tokens_absent(): + 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, - "stop_token_ids": [7], - }, - } + {"sampling_options": {}, "stop_conditions": {"max_tokens": 8}} ) assert "min_new_tokens" not in sampling_params - assert sampling_params["stop_token_ids"] == [7] - assert "ignore_eos" not in sampling_params -def _multimodal_min_tokens_request() -> SglangMultimodalRequest: - return SglangMultimodalRequest( - request=PreprocessedRequest( - token_ids=[1, 2, 3], - stop_conditions=StopConditions(max_tokens=64, min_tokens=64), - sampling_options=SamplingOptions(), - ) - ) - - -def test_multimodal_build_sampling_params_holds_engine_open_for_min_tokens(): - sampling_params = SglangUtils.build_sampling_params( - _multimodal_min_tokens_request(), engine_skips_tokenizer=True +def test_build_sampling_params_forwards_explicit_zero_min_tokens(): + handler = _new_decode_handler( + use_sglang_tokenizer=False, + skip_tokenizer_init=True, ) - # Tokenizer-free engine: Dynamo's decoder enforces the floor, so the engine is only - # kept from stopping early. Over-generation is bounded by max_tokens. - assert "min_new_tokens" not in sampling_params - assert sampling_params["ignore_eos"] is True - assert sampling_params["max_new_tokens"] == 64 - - -def test_multimodal_build_sampling_params_maps_min_tokens_with_engine_tokenizer(): - sampling_params = SglangUtils.build_sampling_params( - _multimodal_min_tokens_request(), engine_skips_tokenizer=False + sampling_params = handler._build_sampling_params( + {"sampling_options": {}, "stop_conditions": {"max_tokens": 8, "min_tokens": 0}} ) - assert sampling_params["min_new_tokens"] == 64 - assert sampling_params["max_new_tokens"] == 64 - assert "ignore_eos" not in sampling_params + assert sampling_params["min_new_tokens"] == 0 -def test_multimodal_build_sampling_params_keeps_an_omitted_ceiling_open(): +def test_multimodal_build_sampling_params_maps_min_tokens(): request = SglangMultimodalRequest( request=PreprocessedRequest( token_ids=[1, 2, 3], - stop_conditions=StopConditions(min_tokens=64), + stop_conditions=StopConditions( + max_tokens=64, min_tokens=64, ignore_eos=True + ), sampling_options=SamplingOptions(), ) ) - sampling_params = SglangUtils.build_sampling_params( - request, engine_skips_tokenizer=True - ) + sampling_params = SglangUtils.build_sampling_params(request) - # A caller that asked only for a floor gets no ceiling. max_new_tokens must be - # present and None -- SGLang reads that as "until EOS or context length", whereas - # omitting the key applies a default of 128 and would cap a larger floor. - assert "max_new_tokens" in sampling_params - assert sampling_params["max_new_tokens"] is None - assert "min_new_tokens" not in sampling_params + assert sampling_params["min_new_tokens"] == 64 + assert sampling_params["max_new_tokens"] == 64 assert sampling_params["ignore_eos"] is True diff --git a/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py b/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py index 360f6e032a0f..3631a96335f5 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py +++ b/components/src/dynamo/sglang/tests/test_sglang_multimodal_video.py @@ -707,7 +707,6 @@ async def test_multimodal_decode_rejects_parallel_sampling_before_prefill_handof """Reject multimodal n greater than one before contacting prefill.""" handler = MultimodalWorkerHandler.__new__(MultimodalWorkerHandler) handler.serving_mode = DisaggregationMode.DECODE - handler.engine_skips_tokenizer = True prefill_called = False class _PrefillClient: diff --git a/lib/llm/src/backend.rs b/lib/llm/src/backend.rs index b9c99627ceb5..1810b23cddb3 100644 --- a/lib/llm/src/backend.rs +++ b/lib/llm/src/backend.rs @@ -253,7 +253,22 @@ impl }); } - match state.stream.next().await { + let output = loop { + let output = state.stream.next().await; + let choice_already_finished = output.as_ref().is_some_and(|output| { + !output.is_event() + && output.data.as_ref().is_some_and(|data| { + state + .finished_choices + .contains(&data.index.unwrap_or_default()) + }) + }); + if !choice_already_finished { + break output; + } + }; + + match output { Some(output) => { // move to state.process_output // handle any error conditions / unwraps here @@ -487,13 +502,9 @@ impl } data.text = text; - // A local stop can fire part-way through a chunk, in which - // case `tokens` holds only the prefix the decoder consumed. - // Every array indexed alongside it has to be cut to the same - // length, or downstream counts token_ids and reports more - // tokens than were emitted, and completion_token_ids carries - // the tokens that came after the stop. `cum_log_probs` is one - // value for the sequence, not per token, so it stays. + // A tokenizer-free engine can return tokens past a local + // stop in the same chunk. Keep its per-token fields aligned + // with the prefix the decoder consumed. let consumed = tokens.len(); if consumed < data.token_ids.len() { data.token_ids.truncate(consumed); @@ -810,9 +821,8 @@ impl Decoder { /// In the future, this method may kick off async cpu/tokio tasks and or async cuda tasks to /// handle logits post-processing and/or other tasks. pub fn step(&mut self, token_id: TokenIdType) -> Result { - // Decided before this token is counted, so a token the floor discards - // does not advance the floor. - let below_min_tokens = self.generated_tokens < self.min_tokens; + // increment the generated tokens + self.generated_tokens += 1; // decode the token let detokenize_start = self.tracker.as_ref().map(|_| Instant::now()); @@ -825,21 +835,10 @@ impl Decoder { } // stop conditions to not apply until the minimum number of tokens have been generated - if below_min_tokens { - // A hidden stop token never appears in the output. Below the floor - // the stop is not honored either, so the token is dropped rather - // than emitted as text — and it does not count toward the minimum, - // which is a floor on the tokens the caller actually receives. - if self.hidden_stop_ids.contains(&token_id) { - return Ok(StepResult::ok(None)); - } - self.generated_tokens += 1; + if self.generated_tokens < self.min_tokens { return Ok(StepResult::ok(token)); } - // increment the generated tokens - self.generated_tokens += 1; - // Check token stops. Visible token IDs are included in output. if self.visible_stop_ids.contains(&token_id) { // Any text still withheld as a partial hidden-stop-sequence match can never @@ -1177,6 +1176,7 @@ mod tests { [102] => " Okay", [103] => "<|user|>", [101, 103] => "Okay<|user|>", + [101, 103, 102] => "Okay<|user|> Okay", _ => anyhow::bail!("unexpected token IDs: {token_ids:?}"), }; Ok(traits::DecodeResult::Complete(token.to_string())) @@ -1199,58 +1199,26 @@ mod tests { &self, request: SingleIn, ) -> Result>, Error> { - let output = LLMEngineOutput { - // SGLang's output_ids_through_stop() includes the matched stop - // position. The frontend must hide its text without mutating - // this raw token sequence. - token_ids: vec![101, 103], - finish_reason: Some(FinishReason::Stop), - index: Some(0), - ..Default::default() - }; - - Ok(ResponseStream::new( - Box::pin(futures::stream::once(async move { - Annotated::from_data(output) - })), - request.context(), - )) - } - } - - struct SyntheticSglangStopWithSuffixEngine; - - #[async_trait] - impl AsyncEngine, ManyOut>, Error> - for SyntheticSglangStopWithSuffixEngine - { - async fn generate( - &self, - request: SingleIn, - ) -> Result>, Error> { - let candidate = |token_id| { - vec![TopLogprob { - rank: 1, - token_id, - token: None, - logprob: -0.5, - bytes: None, - }] - }; - let output = LLMEngineOutput { - // An engine held open past its stop token keeps decoding, so a - // single chunk can carry tokens from after the stop. - token_ids: vec![101, 103, 102], - log_probs: Some(vec![-0.1, -0.2, -0.3]), - top_logprobs: Some(vec![candidate(101), candidate(103), candidate(102)]), - index: Some(0), - ..Default::default() + let output = |index, token_ids: Vec| { + let len = token_ids.len(); + Annotated::from_data(LLMEngineOutput { + token_ids, + log_probs: Some(vec![-0.1; len]), + top_logprobs: Some(vec![vec![]; len]), + index: Some(index), + ..Default::default() + }) }; + let outputs = vec![ + // Choice 0 stops inside this chunk; its suffix must be removed. + output(0, vec![101, 103, 102]), + // It keeps generating while choice 1 is unfinished; drop this chunk. + output(0, vec![102]), + output(1, vec![101, 103]), + ]; Ok(ResponseStream::new( - Box::pin(futures::stream::once(async move { - Annotated::from_data(output) - })), + Box::pin(futures::stream::iter(outputs)), request.context(), )) } @@ -1397,7 +1365,7 @@ mod tests { } #[tokio::test] - async fn test_sglang_hidden_stop_text_is_suppressed_without_mutating_tito_ids() { + async fn test_sglang_local_stop_discards_all_suffix_tokens() { use crate::protocols::common::extensions::NvExt; let tokenizer: Arc = Arc::new(CandidateDecoder); @@ -1409,91 +1377,45 @@ mod tests { stop_token_ids_hidden: Some(vec![103]), ..Default::default() }) - .sampling_options(SamplingOptions::default()) + .sampling_options(SamplingOptions { + n: Some(2), + ..Default::default() + }) .output_options(OutputOptions::default()) .build() .expect("valid preprocessed request"); let engine: ServerStreamingEngine> = Arc::new(SyntheticSglangStopEngine); - let mut stream = Operator::generate(backend.as_ref(), SingleIn::new(request), engine) + let stream = Operator::generate(backend.as_ref(), SingleIn::new(request), engine) .await .expect("backend generation succeeds"); - let output = stream - .next() - .await - .expect("backend emits a response") - .data - .expect("response contains backend output"); - - assert_eq!(output.text.as_deref(), Some("Okay")); - assert_eq!(output.tokens, vec![Some("Okay".to_string()), None]); - assert_eq!(output.token_ids, vec![101, 103]); - assert_eq!(output.finish_reason, Some(FinishReason::Stop)); - - let nvext = NvExt::builder() - .extra_fields(vec!["completion_token_ids".to_string()]) - .build() - .expect("valid nvext selection"); - let options = DeltaGeneratorOptions::new(None, None, false, Some(&nvext)); - let mut generator = DeltaGenerator::new( - "test-model".to_string(), - options, - "test-request".to_string(), - ); - let response = generator - .choice_from_postprocessor(output) - .expect("OpenAI response conversion succeeds"); - + let mut outputs = stream.collect::>().await; assert_eq!( - response - .nvext - .as_ref() - .and_then(|fields| fields.get("completion_token_ids")), - Some(&serde_json::json!([101, 103])) + outputs.len(), + 2, + "chunks after a local stop must be dropped" ); - assert_eq!(generator.get_usage().completion_tokens, 2); - } - - #[tokio::test] - async fn test_local_stop_truncates_every_token_aligned_field() { - use crate::protocols::common::extensions::NvExt; - - let tokenizer: Arc = Arc::new(CandidateDecoder); - let backend = Backend::from_tokenizer(Tokenizer::from(tokenizer)); - let request = PreprocessedRequest::builder() - .model("test-model".to_string()) - .token_ids(vec![]) - .stop_conditions(StopConditions { - stop_token_ids_hidden: Some(vec![103]), - ..Default::default() - }) - .sampling_options(SamplingOptions::default()) - .output_options(OutputOptions { - logprobs: Some(1), - ..Default::default() - }) - .build() - .expect("valid preprocessed request"); - let engine: ServerStreamingEngine> = - Arc::new(SyntheticSglangStopWithSuffixEngine); - - let mut stream = Operator::generate(backend.as_ref(), SingleIn::new(request), engine) - .await - .expect("backend generation succeeds"); - let output = stream - .next() - .await - .expect("backend emits a response") + let output = outputs + .remove(0) + .data + .expect("response contains backend output"); + let second = outputs + .remove(0) .data .expect("response contains backend output"); + assert_eq!(output.index, Some(0)); assert_eq!(output.text.as_deref(), Some("Okay")); assert_eq!(output.tokens, vec![Some("Okay".to_string()), None]); assert_eq!(output.token_ids, vec![101, 103]); - assert_eq!(output.log_probs, Some(vec![-0.1, -0.2])); + assert_eq!(output.log_probs, Some(vec![-0.1; 2])); assert_eq!(output.top_logprobs.as_ref().map(Vec::len), Some(2)); assert_eq!(output.finish_reason, Some(FinishReason::Stop)); + assert_eq!(second.index, Some(1)); + assert_eq!(second.text.as_deref(), Some("Okay")); + assert_eq!(second.token_ids, vec![101, 103]); + assert_eq!(second.finish_reason, Some(FinishReason::Stop)); let nvext = NvExt::builder() .extra_fields(vec!["completion_token_ids".to_string()]) @@ -1514,8 +1436,7 @@ mod tests { .nvext .as_ref() .and_then(|fields| fields.get("completion_token_ids")), - Some(&serde_json::json!([101, 103])), - "the token after the stop must not reach the client" + Some(&serde_json::json!([101, 103])) ); assert_eq!(generator.get_usage().completion_tokens, 2); } diff --git a/lib/llm/tests/test_stop_behavior.rs b/lib/llm/tests/test_stop_behavior.rs index 0f2b7e9d7c5c..5e738b387640 100644 --- a/lib/llm/tests/test_stop_behavior.rs +++ b/lib/llm/tests/test_stop_behavior.rs @@ -145,76 +145,8 @@ fn min_tokens_delays_stop() { let mut decoder = make_decoder(None, Some(3), Some(vec![EOS]), None, false); let result = decoder.process_token_ids(&[HI, EOS]).unwrap(); - assert_eq!(result.text.as_deref(), Some("hi")); - assert!(result.stop_trigger.is_none()); -} - -#[test] -fn min_tokens_stops_at_eos_once_floor_is_reached() { - let mut decoder = make_decoder(None, Some(2), Some(vec![EOS]), None, false); - let result = decoder.process_token_ids(&[HI, THERE, EOS]).unwrap(); - - assert_eq!(result.text.as_deref(), Some("hithere")); - assert!(matches!( - result.stop_trigger, - Some(StopTrigger::HiddenStopTokenDetected(id)) if id == EOS - )); -} - -/// The floor is a floor on the tokens the caller receives. An end-of-sequence -/// token arriving on the boundary is not one of them, so with a floor of two it -/// neither stops the sequence nor counts toward the two. -#[test] -fn min_tokens_does_not_count_a_suppressed_eos() { - let mut decoder = make_decoder(None, Some(2), Some(vec![EOS]), None, false); - let result = decoder.process_token_ids(&[HI, EOS]).unwrap(); - - assert_eq!(result.text.as_deref(), Some("hi")); - assert!(result.stop_trigger.is_none()); - - let result = decoder.process_token_ids(&[THERE, EOS]).unwrap(); - assert_eq!(result.text.as_deref(), Some("there")); - assert!(matches!( - result.stop_trigger, - Some(StopTrigger::HiddenStopTokenDetected(id)) if id == EOS - )); -} - -/// A suppressed token still occupies its slot in `tokens`, so the arrays the -/// backend indexes alongside it stay aligned with the token ids it was given. -#[test] -fn min_tokens_keeps_suppressed_eos_aligned() { - let mut decoder = make_decoder(None, Some(3), Some(vec![EOS]), None, false); - let result = decoder.process_token_ids(&[HI, EOS, THERE]).unwrap(); - - assert_eq!(result.tokens.len(), 3); - assert_eq!(result.tokens[1], None); - assert_eq!(result.text.as_deref(), Some("hithere")); -} - -/// A visible stop token is part of the output, so below the floor it is emitted -/// and counted like any other token — only the stop itself waits. -#[test] -fn min_tokens_keeps_visible_stop_tokens_below_the_floor() { - let tokenizer: Arc = Arc::new(TestTokenizer); - let decode_stream = tokenizers::DecodeStream::new(tokenizer, &[], false); - let stop_conditions = StopConditions { - min_tokens: Some(3), - stop_token_ids_visible: Some(vec![STOP]), - ..Default::default() - }; - let mut decoder = Decoder::new(decode_stream, stop_conditions, false, None, None); - - let result = decoder.process_token_ids(&[HI, STOP]).unwrap(); - assert_eq!(result.text.as_deref(), Some("hiSTOP")); + assert_eq!(result.text.as_deref(), Some("hi")); assert!(result.stop_trigger.is_none()); - - let result = decoder.process_token_ids(&[THERE, STOP]).unwrap(); - assert_eq!(result.text.as_deref(), Some("thereSTOP")); - assert!(matches!( - result.stop_trigger, - Some(StopTrigger::VisibleStopTokenDetected(id)) if id == STOP - )); } #[test] diff --git a/lib/sidecar/sglang/src/native_http.rs b/lib/sidecar/sglang/src/native_http.rs index f4e3127ed394..5c1000fb9f63 100644 --- a/lib/sidecar/sglang/src/native_http.rs +++ b/lib/sidecar/sglang/src/native_http.rs @@ -47,11 +47,6 @@ pub(crate) fn request( else { return Ok(None); }; - if !mode.is_prefill() && request.stop_conditions.min_tokens.unwrap_or(0) > 0 { - // Native HTTP responses bypass Dynamo's Decoder, which owns minimum-token - // enforcement when SGLang runs without a tokenizer. - return Ok(None); - } let mut body = payload .as_object() .cloned() diff --git a/lib/sidecar/sglang/src/protocol.rs b/lib/sidecar/sglang/src/protocol.rs index db6ea9bf697b..b9afe744c5fb 100644 --- a/lib/sidecar/sglang/src/protocol.rs +++ b/lib/sidecar/sglang/src/protocol.rs @@ -40,10 +40,16 @@ pub(crate) fn build_generate_request( .transpose() .map_err(|_| client::invalid_arg("max_tokens does not fit in i32"))? }; - // SGLang rejects `min_new_tokens` without a tokenizer and this path never has one, so the - // `Decoder` in `lib/llm/src/backend.rs` enforces the floor and the engine is held open. - let hold_engine_open = - !mode.is_prefill() && request.stop_conditions.min_tokens.unwrap_or(0) > 0; + let min_new_tokens = if mode.is_prefill() { + None + } else { + request + .stop_conditions + .min_tokens + .map(i32::try_from) + .transpose() + .map_err(|_| client::invalid_arg("min_tokens does not fit in i32"))? + }; let mut stop_token_ids = Vec::new(); for tokens in [ @@ -62,13 +68,6 @@ pub(crate) fn build_generate_request( } } } - let mut stop = request.stop_conditions.stop.clone().unwrap_or_default(); - if hold_engine_open { - // The `Decoder` withholds every stop until the floor is reached, stop strings - // included, so any stop left with the engine can end the stream below the floor. - stop_token_ids.clear(); - stop.clear(); - } let guided = request.sampling_options.guided_decoding.as_ref(); let sampling_params = pb::SamplingParams { @@ -80,14 +79,10 @@ pub(crate) fn build_generate_request( presence_penalty: request.sampling_options.presence_penalty, repetition_penalty: request.sampling_options.repetition_penalty, max_new_tokens, - min_new_tokens: None, - stop, + min_new_tokens, + stop: request.stop_conditions.stop.clone().unwrap_or_default(), stop_token_ids, - ignore_eos: if hold_engine_open { - Some(true) - } else { - request.stop_conditions.ignore_eos - }, + ignore_eos: request.stop_conditions.ignore_eos, n: request.sampling_options.n.map(i32::from), json_schema: guided .and_then(|value| value.json.as_ref()) @@ -654,54 +649,6 @@ mod tests { assert_eq!(mapped.disaggregated_params.unwrap().bootstrap_port, 5001); } - #[test] - fn decode_min_tokens_is_left_to_the_frontend_and_engine_is_held_open() { - let mut request = request(); - request.stop_conditions.min_tokens = Some(4); - request.stop_conditions.stop_token_ids = Some(vec![7]); - request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); - request.stop_conditions.stop = Some(vec!["STOP".to_string()]); - - let mapped = build_generate_request( - &request, - "rid-min", - DisaggregationMode::Aggregated, - None, - None, - ) - .unwrap(); - let sampling = mapped.sampling_params.unwrap(); - - assert_eq!(sampling.min_new_tokens, None); - assert!(sampling.stop_token_ids.is_empty()); - assert!(sampling.stop.is_empty()); - assert_eq!(sampling.ignore_eos, Some(true)); - assert_eq!(sampling.max_new_tokens, Some(8)); - } - - #[test] - fn decode_without_min_tokens_keeps_engine_side_stopping() { - let mut request = request(); - request.stop_conditions.stop_token_ids = Some(vec![7]); - request.stop_conditions.stop_token_ids_hidden = Some(vec![151643]); - request.stop_conditions.stop = Some(vec!["STOP".to_string()]); - - let mapped = build_generate_request( - &request, - "rid-no-min", - DisaggregationMode::Aggregated, - None, - None, - ) - .unwrap(); - let sampling = mapped.sampling_params.unwrap(); - - assert_eq!(sampling.min_new_tokens, None); - assert_eq!(sampling.stop_token_ids, vec![7, 151643]); - assert_eq!(sampling.stop, vec!["STOP".to_string()]); - assert_eq!(sampling.ignore_eos, None); - } - #[test] fn prefill_uses_selected_prefill_dp_rank() { let mut request = request(); From 3c8797cef134b29a200f1eef1d8d44a4e3f84d3b Mon Sep 17 00:00:00 2001 From: GLAMR Date: Tue, 15 Sep 2026 21:23:35 +0000 Subject: [PATCH 10/15] ci: supply required framework for Triton tests Signed-off-by: GLAMR --- .github/workflows/pr.yaml | 1 + 1 file changed, 1 insertion(+) diff --git a/.github/workflows/pr.yaml b/.github/workflows/pr.yaml index 57936e0b4ed6..09428070b404 100644 --- a/.github/workflows/pr.yaml +++ b/.github/workflows/pr.yaml @@ -940,6 +940,7 @@ jobs: uses: $/.github/workflows/shared-test.yml with: test_suite_name: triton + framework: triton test_type: CPU Test amd_runner: prod-tester-amd-v2 target_tag_plain: ${{ needs.triton-build.outputs.target_tag_plain }} From e2941f956882067780fdf352fdfbf16fb78472c6 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Tue, 15 Sep 2026 22:36:17 +0000 Subject: [PATCH 11/15] fix(sglang): honor minimum boundary and abort all choices Signed-off-by: GLAMR --- .../sglang/request_handlers/handler_base.py | 46 +++++++++++-------- .../request_handlers/llm/decode_handler.py | 17 +++++-- lib/llm/src/backend.rs | 3 +- lib/llm/tests/test_stop_behavior.rs | 8 +++- 4 files changed, 48 insertions(+), 26 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index 26cf4e1fd0e8..b0e76d16940d 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -1234,8 +1234,21 @@ def _get_bootstrap_info(engine: sgl.Engine) -> Tuple[str, int]: return bootstrap_host, bootstrap_port + def _abort_requests(self, request_ids: set[str]) -> 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() + 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. @@ -1243,6 +1256,7 @@ async def _handle_cancellation( 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. @@ -1288,22 +1302,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} + ) # Check which event triggered and raise EngineShutdown if shutdown if shutdown_task and shutdown_task in done: @@ -1333,7 +1334,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. @@ -1346,6 +1350,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 @@ -1354,12 +1359,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) # Clean up the background cancellation task request_id = "unknown" if request_id_future.done() and not request_id_future.cancelled(): diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index ab184517070c..dc4e3acdda33 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -707,16 +707,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) + continue # Check cancellation before yielding to allow proper cleanup. # This lets SGLang proceed to the second token generation, which will diff --git a/lib/llm/src/backend.rs b/lib/llm/src/backend.rs index 1810b23cddb3..bb57db033193 100644 --- a/lib/llm/src/backend.rs +++ b/lib/llm/src/backend.rs @@ -821,6 +821,7 @@ impl Decoder { /// In the future, this method may kick off async cpu/tokio tasks and or async cuda tasks to /// handle logits post-processing and/or other tasks. pub fn step(&mut self, token_id: TokenIdType) -> Result { + let below_min_tokens = self.generated_tokens < self.min_tokens; // increment the generated tokens self.generated_tokens += 1; @@ -835,7 +836,7 @@ impl Decoder { } // stop conditions to not apply until the minimum number of tokens have been generated - if self.generated_tokens < self.min_tokens { + if below_min_tokens { return Ok(StepResult::ok(token)); } diff --git a/lib/llm/tests/test_stop_behavior.rs b/lib/llm/tests/test_stop_behavior.rs index 5e738b387640..47825ceff85b 100644 --- a/lib/llm/tests/test_stop_behavior.rs +++ b/lib/llm/tests/test_stop_behavior.rs @@ -142,11 +142,17 @@ fn trailing_tokens_ignored_after_stop() { #[test] fn min_tokens_delays_stop() { - let mut decoder = make_decoder(None, Some(3), Some(vec![EOS]), None, false); + let mut decoder = make_decoder(None, Some(2), Some(vec![EOS]), None, false); let result = decoder.process_token_ids(&[HI, EOS]).unwrap(); assert_eq!(result.text.as_deref(), Some("hi")); assert!(result.stop_trigger.is_none()); + + let result = decoder.process_token_ids(&[EOS]).unwrap(); + assert!(matches!( + result.stop_trigger, + Some(StopTrigger::HiddenStopTokenDetected(id)) if id == EOS + )); } #[test] From f0ac9f34a5f7911dd9bd1a97dbed1a0874f38e4f Mon Sep 17 00:00:00 2001 From: GLAMR Date: Wed, 16 Sep 2026 18:28:40 +0000 Subject: [PATCH 12/15] ci: declare networkx for test collection Signed-off-by: GLAMR --- container/deps/requirements.test.txt | 2 ++ 1 file changed, 2 insertions(+) diff --git a/container/deps/requirements.test.txt b/container/deps/requirements.test.txt index 15a6091a0809..a9153b3848da 100644 --- a/container/deps/requirements.test.txt +++ b/container/deps/requirements.test.txt @@ -45,6 +45,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 # Protobuf definitions the OTLP parity receiver decodes; see grpcio above. # Version-matched to the opentelemetry-* pins in requirements.common.txt. From 7ca35cee32f252bf6c1f42886ee990d3db2ae683 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Wed, 16 Sep 2026 18:47:51 +0000 Subject: [PATCH 13/15] test(sglang): provide stop state in shutdown context stub Signed-off-by: GLAMR --- .../src/dynamo/sglang/tests/test_sglang_decode_handler.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py index 80619b57b19d..67dabd20773b 100644 --- a/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py +++ b/components/src/dynamo/sglang/tests/test_sglang_decode_handler.py @@ -276,7 +276,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 { From c66ce8a8520bf2d6a315166f300ff606d9cf4f64 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Wed, 16 Sep 2026 20:08:43 +0000 Subject: [PATCH 14/15] fix(sglang): retain context logging after request aborts Signed-off-by: GLAMR --- .../src/dynamo/sglang/request_handlers/handler_base.py | 7 ++++--- .../dynamo/sglang/request_handlers/llm/decode_handler.py | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/handler_base.py b/components/src/dynamo/sglang/request_handlers/handler_base.py index b0e76d16940d..d756d1f9a999 100644 --- a/components/src/dynamo/sglang/request_handlers/handler_base.py +++ b/components/src/dynamo/sglang/request_handlers/handler_base.py @@ -1234,7 +1234,7 @@ def _get_bootstrap_info(engine: sgl.Engine) -> Tuple[str, int]: return bootstrap_host, bootstrap_port - def _abort_requests(self, request_ids: set[str]) -> None: + def _abort_requests(self, request_ids: set[str], context: Context) -> None: if not request_ids: return if not getattr(self.engine, "tokenizer_manager", None): @@ -1243,6 +1243,7 @@ def _abort_requests(self, request_ids: set[str]) -> None: 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, @@ -1303,7 +1304,7 @@ async def _handle_cancellation( ) self._abort_requests( - request_ids if request_ids is not None else {sglang_request_id} + request_ids if request_ids is not None else {sglang_request_id}, context ) # Check which event triggered and raise EngineShutdown if shutdown @@ -1367,7 +1368,7 @@ async def _cancellation_monitor( finally: # Stream closure can beat the monitor's cancellation wake-up. if request_ids is not None: - self._abort_requests(request_ids) + 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(): diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index dc4e3acdda33..0e27d37af675 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -722,7 +722,7 @@ async def _process_token_stream( 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) + self._abort_requests(request_ids, context) continue # Check cancellation before yielding to allow proper cleanup. From 21638df03d461b6f744f45b28ca95195120f8de7 Mon Sep 17 00:00:00 2001 From: GLAMR Date: Wed, 16 Sep 2026 20:23:35 +0000 Subject: [PATCH 15/15] fix(sglang): abort every unfinished text-stream choice Signed-off-by: GLAMR --- .../request_handlers/llm/decode_handler.py | 16 +++++++++++----- 1 file changed, 11 insertions(+), 5 deletions(-) diff --git a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py index 0e27d37af675..02fbd373e01e 100644 --- a/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py +++ b/components/src/dynamo/sglang/request_handlers/llm/decode_handler.py @@ -852,16 +852,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