Rust server: align launcher and request validation behavior (#37327)

python/sglang/srt/managers/scheduler.py

@@ -2171,6 +2171,9 @@ class Scheduler(
         # Park the idle loop on the request ring within the rank-0 rust-server
         self.idle_sleeper = RustServerIdleSleeper(rust_server)
 
+    def rust_server_tokenizer_path(self) -> str:
+        return get_serving().tokenizer_path
+
     def init_request_receiver(self) -> None:
         self.request_receiver = SchedulerRequestReceiver(
             recv_from_tokenizer=self.recv_from_tokenizer,

python/sglang/srt/rust_server/config.py

@@ -46,7 +46,7 @@ def _build_server_args(scheduler: Scheduler) -> ServerArgs:
     return ext.ServerArgs(
         model_path=get_model().model_path,
         served_model_name=get_serving().served_model_name,
-        tokenizer_path=get_serving().tokenizer_path,
+        tokenizer_path=scheduler.rust_server_tokenizer_path(),
         revision=get_model().revision,
         load_format=get_model().load_format,
         weight_version=get_serving().weight_version,
@@ -76,8 +76,6 @@ def _build_server_args(scheduler: Scheduler) -> ServerArgs:
                 **mc.get_default_sampling_params()
             ),
         ),
-        # `preferred_sampling_params` is deliberately absent: `launch`
-        # refuses to start when it is set, so the Rust server never needs it.
         preferred_sampling_params=(
             json.dumps(get_serving().preferred_sampling_params)
             if get_serving().preferred_sampling_params is not None

python/sglang/srt/rust_server/server.py

@@ -35,6 +35,7 @@ from sglang.srt.utils.flatten import (
     NestedRowColumns,
     RaggedPairColumns,
 )
+from sglang.srt.utils.network import NetworkAddress
 
 if TYPE_CHECKING:
     from sglang.srt.managers.io_struct import BatchTokenIDOutput
@@ -76,24 +77,11 @@ class RustServer:
         os.environ.setdefault("TOKENIZERS_PARALLELISM", "false")
 
         server_args = scheduler.server_args
-        # `TokenizerManager` merges these under each request's own sampling params
-        # (`{**preferred, **obj.sampling_params}`), and this server replaces that
-        # manager wholesale — so honouring the flag is not implemented here yet.
-        # Refuse rather than run: silently dropping it means generating with
-        # sampling the operator did not configure, and `/get_model_info` would go on
-        # advertising values no request ever receives.
-        if get_serving().preferred_sampling_params:
-            raise ValueError(
-                "SGLANG_RUST_SERVER does not yet apply --preferred-sampling-params "
-                "(the Python TokenizerManager merges it into every request; the rust "
-                "ingress has no equivalent). Launch without SGLANG_RUST_SERVER, or "
-                "drop --preferred-sampling-params and send those values per request."
-            )
         # Per-DP-rank HTTP port with client load balancing. `None` when DP is off,
         # so the rank is not conflated with rank 0 of a one-rank group.
         dp_rank = scheduler.ps.attn_dp_rank if scheduler.ps.dp_size > 1 else None
         listen_port = get_serving().port + (dp_rank or 0)
-        listen_addr = f"{get_serving().host}:{listen_port}"
+        listen_addr = NetworkAddress(get_serving().host, listen_port).to_host_port_str()
 
         launch_cores, server_cores = _partition_cores(
             mm_workers=(

rust/sglang-server/src/api_server/native_api.rs

@@ -196,7 +196,7 @@ async fn generate(
     State(state): State<Arc<AppState>>,
     body: Result<Json<GenerateBody>, JsonRejection>,
 ) -> Response {
-    let body = match body {
+    let mut body = match body {
         Ok(Json(body)) => body,
         // A body that fails to parse has no readable `stream` flag, so this one
         // can only answer unary — as Python's does (FastAPI rejects before its
@@ -206,6 +206,15 @@ async fn generate(
         }
     };
     let stream = body.stream;
+    if let Some(preferred) = &state.server_args.preferred_sampling_params
+        && let Err(error) = body.apply_preferred_sampling(&preferred.0)
+    {
+        return native_error(
+            StatusCode::INTERNAL_SERVER_ERROR,
+            &error.to_string(),
+            stream,
+        );
+    }
     // Fan `text`/`input_ids`/`sampling_params` (scalar or list) into per-request
     // payloads. `is_batch` = list form → the response is a JSON array.
     let (mut payloads, is_batch) = match body.into_requests() {

rust/sglang-server/src/message/config.rs

@@ -126,8 +126,8 @@ pub struct ServerArgs {
     pub disaggregation_mode: DisaggregationMode,
     /// The resolved Python `ModelConfig`, attached at handoff time.
     pub model_config: ModelConfig,
-    /// Default sampling params advertised by `/get_model_info`, verbatim from
-    /// `server_args.preferred_sampling_params` (a JSON object or null).
+    /// Launch-time sampling defaults merged beneath per-request values and
+    /// advertised by `/get_model_info`.
     pub preferred_sampling_params: Option<PreferredSamplingParams>,
     /// Over-long inputs are truncated to fit the context instead of 400ing, and
     /// `max_new_tokens` is clamped rather than rejected (Python
@@ -528,6 +528,10 @@ impl ServerArgs {
         if self.served_model_name.is_empty() {
             return Err("empty 'served_model_name' in server_args".into());
         }
+        if let Some(preferred) = &self.preferred_sampling_params {
+            super::sampling::SamplingParamsInput::from_preferred(&preferred.0)
+                .map_err(|e| format!("invalid preferred_sampling_params: {e}"))?;
+        }
         Ok(())
     }
 

rust/sglang-server/src/message/request.rs

@@ -109,6 +109,18 @@ pub struct GenerateBody {
 }
 
 impl GenerateBody {
+    /// Merge operator-provided sampling defaults beneath request values,
+    /// matching Python TokenizerManager's preferred/request precedence.
+    pub fn apply_preferred_sampling(&mut self, preferred: &serde_json::Value) -> Result<(), Error> {
+        match &mut self.sampling_params {
+            Some(params) => params.apply_preferred(preferred),
+            None => SamplingParamsInput::from_preferred(preferred).map(|params| {
+                self.sampling_params = Some(params);
+            }),
+        }
+        .map_err(|e| Error::Validation(format!("invalid preferred_sampling_params: {e}")))
+    }
+
     /// Validate, normalize and fan the body into one [`GenerateRequest`] per
     /// prompt + `is_batch` (list form — a 1-element list is still a batch → JSON
     /// array response). The Rust counterpart of Python

rust/sglang-server/src/message/sampling.rs

@@ -3,7 +3,7 @@
 //! `__post_init__` → `normalize` → `verify` pipeline (run in that order, as
 //! `TokenizerManager._create_tokenized_object` does).
 
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::fmt;
 
 use serde::de::value::{MapAccessDeserializer, SeqAccessDeserializer};
@@ -233,6 +233,11 @@ pub struct SamplingParams {
     /// Set by `normalize`; tells the scheduler its own pass can early-return.
     #[serde(skip_deserializing)]
     pub is_normalized: bool,
+    /// API fields present in the request object. Serde defaults erase this
+    /// distinction, but preferred sampling parameters must not overwrite an
+    /// explicit request value, including an explicit default or null.
+    #[serde(skip)]
+    pub(crate) explicit_fields: BTreeSet<String>,
 }
 
 /// The `/generate` body's `sampling_params`: one object (broadcast to every
@@ -263,12 +268,21 @@ impl<'de> Deserialize<'de> for SamplingParamsInput {
             }
 
             fn visit_map<A: MapAccess<'de>>(self, map: A) -> Result<Self::Value, A::Error> {
-                SamplingParams::deserialize(MapAccessDeserializer::new(map))
+                let value = serde_json::Value::deserialize(MapAccessDeserializer::new(map))?;
+                sampling_params_from_value(value)
                     .map(|p| SamplingParamsInput::One(Box::new(p)))
+                    .map_err(serde::de::Error::custom)
             }
 
             fn visit_seq<A: SeqAccess<'de>>(self, seq: A) -> Result<Self::Value, A::Error> {
-                Vec::deserialize(SeqAccessDeserializer::new(seq)).map(SamplingParamsInput::Many)
+                let values =
+                    Vec::<serde_json::Value>::deserialize(SeqAccessDeserializer::new(seq))?;
+                values
+                    .into_iter()
+                    .map(sampling_params_from_value)
+                    .collect::<Result<Vec<_>, _>>()
+                    .map(SamplingParamsInput::Many)
+                    .map_err(serde::de::Error::custom)
             }
         }
 
@@ -276,6 +290,56 @@ impl<'de> Deserialize<'de> for SamplingParamsInput {
     }
 }
 
+fn sampling_params_from_value(value: serde_json::Value) -> Result<SamplingParams, String> {
+    let explicit_fields = value
+        .as_object()
+        .ok_or_else(|| "sampling_params must be an object".to_string())?
+        .keys()
+        .cloned()
+        .collect();
+    let mut params: SamplingParams = serde_json::from_value(value).map_err(|e| e.to_string())?;
+    params.explicit_fields = explicit_fields;
+    Ok(params)
+}
+
+impl SamplingParamsInput {
+    /// Merge launch-time preferred params beneath request params. A request key
+    /// wins even when it explicitly carries the type's default or null.
+    pub fn apply_preferred(&mut self, preferred: &serde_json::Value) -> Result<(), String> {
+        match self {
+            Self::One(params) => apply_preferred_to_one(params, preferred),
+            Self::Many(params) => params
+                .iter_mut()
+                .try_for_each(|params| apply_preferred_to_one(params, preferred)),
+        }
+    }
+
+    pub fn from_preferred(preferred: &serde_json::Value) -> Result<Self, String> {
+        sampling_params_from_value(preferred.clone()).map(|params| Self::One(Box::new(params)))
+    }
+}
+
+fn apply_preferred_to_one(
+    params: &mut SamplingParams,
+    preferred: &serde_json::Value,
+) -> Result<(), String> {
+    let mut merged = preferred
+        .as_object()
+        .ok_or_else(|| "preferred_sampling_params must be a JSON object".to_string())?
+        .clone();
+    let request_value = serde_json::to_value(&*params).map_err(|e| e.to_string())?;
+    let request = request_value
+        .as_object()
+        .ok_or_else(|| "SamplingParams did not serialize as an object".to_string())?;
+    for field in &params.explicit_fields {
+        if let Some(value) = request.get(field) {
+            merged.insert(field.clone(), value.clone());
+        }
+    }
+    *params = sampling_params_from_value(serde_json::Value::Object(merged))?;
+    Ok(())
+}
+
 impl Default for SamplingParams {
     fn default() -> Self {
         // Each field reads the same `default()` the serde attribute above names,
@@ -314,6 +378,7 @@ impl Default for SamplingParams {
             stop_str_max_len: 0,
             stop_regex_max_len: 0,
             is_normalized: false,
+            explicit_fields: BTreeSet::new(),
         }
     }
 }
@@ -1085,4 +1150,35 @@ mod tests {
         let err = norm_err(&json).to_string();
         assert!(err.contains("at most"), "{err}");
     }
+
+    #[test]
+    fn preferred_params_fill_only_omitted_request_fields() {
+        let preferred = serde_json::json!({
+            "temperature": 0.25,
+            "top_p": 0.75,
+            "max_new_tokens": 4096
+        });
+        let mut input: SamplingParamsInput =
+            serde_json::from_str(r#"{"temperature": 1.0, "top_p": null}"#).unwrap();
+        input.apply_preferred(&preferred).unwrap();
+        let SamplingParamsInput::One(params) = input else {
+            panic!("expected scalar params")
+        };
+        assert_eq!(params.temperature, 1.0, "explicit default wins");
+        assert_eq!(params.top_p, 1.0, "explicit null keeps the type default");
+        assert_eq!(params.max_new_tokens, Some(4096), "omitted uses preferred");
+    }
+
+    #[test]
+    fn preferred_params_apply_to_every_batched_object() {
+        let preferred = serde_json::json!({"temperature": 0.25, "top_p": 0.75});
+        let mut input: SamplingParamsInput =
+            serde_json::from_str(r#"[{"temperature": 0.5}, {"top_p": 0.9}]"#).unwrap();
+        input.apply_preferred(&preferred).unwrap();
+        let SamplingParamsInput::Many(params) = input else {
+            panic!("expected batched params")
+        };
+        assert_eq!((params[0].temperature, params[0].top_p), (0.5, 0.75));
+        assert_eq!((params[1].temperature, params[1].top_p), (0.25, 0.9));
+    }
 }

rust/sglang-server/src/tokenizer_manager/to_scheduler.rs

@@ -12,7 +12,9 @@ use crate::message::response::ResponseItem;
 use crate::runtime::Runnable;
 use crate::tokenizer_manager::channel::ToSchedulerTx;
 pub use crate::tokenizer_manager::to_scheduler_types::{Limits, Mm};
-use crate::tokenizer_manager::to_scheduler_validation::{check_total_tokens, validate};
+use crate::tokenizer_manager::to_scheduler_validation::{
+    check_total_tokens, validate, validate_input_ids,
+};
 use crate::tokenizer_manager::wiring::{AbortSource, Senders, TmEvent};
 use crate::utils::{
     error::Error,
@@ -274,7 +276,8 @@ impl Intake {
                 // a text request has no ids yet.
                 RequestState::PreSendValidating => {
                     if let RequestKind::Generate(g) = &mut req.kind
-                        && let Err(e) = check_total_tokens(g, &self.limits)
+                        && let Err(e) = validate_input_ids(g, self.limits.vocab_size)
+                            .and_then(|()| check_total_tokens(g, &self.limits))
                     {
                         let _ = req.state.apply(Event::Error(e)); // → Failed
                         continue;

rust/sglang-server/src/tokenizer_manager/to_scheduler_tests.rs

@@ -636,6 +636,27 @@ fn negative_and_logprob_token_ids_rejected() {
     }
 }
 
+#[test]
+fn multimodal_sentinel_is_validated_after_expansion() {
+    let mut req = generate_req(24, SamplingParams::default());
+    let RequestKind::Generate(g) = &mut req.kind else {
+        unreachable!()
+    };
+    g.input_ids = Some(vec![1, -103, 2]);
+    g.mm = Some(Box::new(crate::message::request::MmData {
+        audio_data: Some(rmpv::Value::from("data:audio/wav;base64,xxxx")),
+        ..Default::default()
+    }));
+
+    assert!(validate(&mut req, &test_limits()).is_ok());
+    let RequestKind::Generate(g) = &mut req.kind else {
+        unreachable!()
+    };
+    assert!(validate_input_ids(g, test_limits().vocab_size).is_err());
+    g.input_ids = Some(vec![1, 103, 2]);
+    assert!(validate_input_ids(g, test_limits().vocab_size).is_ok());
+}
+
 /// A valid request is registered and handed onward — never deregistered.
 #[test]
 fn admitted_request_keeps_registration() {

rust/sglang-server/src/tokenizer_manager/to_scheduler_validation.rs

@@ -41,19 +41,12 @@ pub(super) fn validate(req: &mut Request, limits: &Limits) -> Result<(), Error>
         ));
     }
 
-    // Client-supplied token ids must be in-vocabulary: an out-of-range id
-    // reaches the embedding lookup and kills the scheduler process, so 400
-    // here instead — mirroring the Python `TokenizerManager` validation.
+    // Multimodal processors may consume sentinel ids outside the vocabulary.
+    // Validate their resulting ids at PreSendValidating instead. Non-MM client
+    // ids can be rejected now.
     if let RequestKind::Generate(g) = &req.kind {
-        if let Some(ids) = &g.input_ids {
-            for &id in ids {
-                if id < 0 || id as u64 >= vocab_size {
-                    return Err(Error::Validation(format!(
-                        "input_ids contains out-of-vocabulary token id {id}; \
-                         valid range is [0, {vocab_size})"
-                    )));
-                }
-            }
+        if !g.has_multimodal() {
+            validate_input_ids(g, vocab_size)?;
         }
         if let Some(ids) = &g.token_ids_logprob {
             for &id in ids {
@@ -95,6 +88,22 @@ pub(super) fn validate(req: &mut Request, limits: &Limits) -> Result<(), Error>
     Ok(())
 }
 
+/// Guard the ids that will reach the embedding lookup. Multimodal requests run
+/// this after placeholder expansion; all other requests also run it at intake.
+pub(super) fn validate_input_ids(g: &GenerateRequest, vocab_size: u64) -> Result<(), Error> {
+    if let Some(ids) = &g.input_ids {
+        for &id in ids {
+            if id < 0 || id as u64 >= vocab_size {
+                return Err(Error::Validation(format!(
+                    "input_ids contains out-of-vocabulary token id {id}; \
+                     valid range is [0, {vocab_size})"
+                )));
+            }
+        }
+    }
+    Ok(())
+}
+
 /// The context-window checks that need the tokenized length, mirroring Python
 /// `TokenizerManager._validate_one_request`: the input alone must fit, and then
 /// input + `max_new_tokens` must fit. Without them the scheduler silently clamps

scripts/ci/rerun_test_groups.json

@@ -0,0 +1,8 @@
+{
+  "rust-server": [
+    "registered/rust/test_run_rust_tests.py",
+    "registered/core/test_srt_endpoint.py",
+    "registered/vlm/test_rust_native_mm_e2e.py",
+    "registered/vlm/test_rust_native_mm_mmmu.py"
+  ]
+}

scripts/ci/utils/slash_command_handler.py

@@ -27,6 +27,7 @@ _ALLOWED_INSTALL_SCRIPT = re.compile(r"^scripts/ci/cuda/[\w.-]+\.sh$")
 
 # Configuration
 PERMISSIONS_FILE_PATH = ".github/CI_PERMISSIONS.json"
+TEST_GROUPS_FILE_PATH = "scripts/ci/rerun_test_groups.json"
 PRECISION_BASELINE_TEST = "registered/debug_utils/test_nightly_precision_regression.py"
 PRECISION_BASELINE_REFRESH_FLAG = "--refresh-precision-baseline"
 
@@ -502,11 +503,16 @@ MULTIMODAL_PATH_TO_RUNNER = {
 MULTIMODAL_DEFAULT_RUNNER = "1-gpu-h100"
 
 
+def _load_test_groups():
+    with open(TEST_GROUPS_FILE_PATH, encoding="utf-8") as f:
+        return json.load(f)
+
+
 def _known_test_groups():
-    groups = []
+    groups = set(_load_test_groups())
     for group_dir in glob.glob("test/registered/*"):
         if os.path.isdir(group_dir):
-            groups.append(os.path.basename(group_dir))
+            groups.add(os.path.basename(group_dir))
     return sorted(groups)
 
 
@@ -514,8 +520,9 @@ def resolve_test_group_specs(group_name):
     """
     Resolve a test group name into /rerun-test specs.
 
-    A group maps to a directory under test/registered/. For example,
-    "hicache" maps to all test_*.py files under test/registered/hicache/.
+    A group maps to either a named cross-directory file set or a directory
+    under test/registered/. For example, "hicache" maps to all test_*.py
+    files under test/registered/hicache/.
 
     Returns (test_specs, error_message). On success error_message is None.
     """
@@ -528,6 +535,25 @@ def resolve_test_group_specs(group_name):
     ):
         return [], f"Invalid test group `{group_name}`."
 
+    test_groups = _load_test_groups()
+    if group_name in test_groups:
+        test_specs = test_groups[group_name]
+        if not isinstance(test_specs, list) or not all(
+            isinstance(test_spec, str) for test_spec in test_specs
+        ):
+            return [], f"Invalid definition for test group `{group_name}`."
+        missing = [
+            test_spec
+            for test_spec in test_specs
+            if not os.path.isfile(os.path.join("test", test_spec))
+        ]
+        if missing:
+            return [], (
+                f"Named test group `{group_name}` references missing files: "
+                + ", ".join(f"`test/{path}`" for path in missing)
+            )
+        return test_specs, None
+
     group_dir = os.path.join("test", "registered", group_name)
     if not os.path.isdir(group_dir):
         known = ", ".join(f"`{g}`" for g in _known_test_groups())

test/registered/unit/tools/test_slash_command_handler.py

@@ -0,0 +1,99 @@
+"""Tests for declarative slash-command test groups."""
+
+import importlib.util
+import json
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from types import ModuleType
+from unittest.mock import patch
+
+from sglang.test.ci.ci_register import register_cpu_ci
+from sglang.test.test_utils import CustomTestCase
+
+register_cpu_ci(est_time=1, suite="base-a-test-cpu")
+
+_REPO_ROOT = Path(__file__).resolve().parents[4]
+_HANDLER_PATH = _REPO_ROOT / "scripts/ci/utils/slash_command_handler.py"
+
+
+def _load_handler():
+    github = ModuleType("github")
+    github.Auth = object()
+    github.Github = object()
+    spec = importlib.util.spec_from_file_location(
+        "slash_command_handler", _HANDLER_PATH
+    )
+    module = importlib.util.module_from_spec(spec)
+    with patch.dict(sys.modules, {"github": github}):
+        spec.loader.exec_module(module)
+    return module
+
+
+class TestConfiguredTestGroups(CustomTestCase):
+    def test_additional_group_requires_only_manifest_data(self):
+        handler = _load_handler()
+        previous_cwd = os.getcwd()
+        try:
+            os.chdir(_REPO_ROOT)
+            with tempfile.TemporaryDirectory() as temp_dir:
+                manifest = Path(temp_dir) / "groups.json"
+                manifest.write_text(
+                    json.dumps(
+                        {
+                            "mixed": [
+                                "registered/rust/test_run_rust_tests.py",
+                                "registered/core/test_srt_endpoint.py",
+                            ]
+                        }
+                    )
+                )
+                with patch.object(handler, "TEST_GROUPS_FILE_PATH", str(manifest)):
+                    specs, error = handler.resolve_test_group_specs("mixed")
+
+            self.assertIsNone(error)
+            self.assertEqual(
+                specs,
+                [
+                    "registered/rust/test_run_rust_tests.py",
+                    "registered/core/test_srt_endpoint.py",
+                ],
+            )
+        finally:
+            os.chdir(previous_cwd)
+
+    def test_rust_server_group(self):
+        handler = _load_handler()
+        previous_cwd = os.getcwd()
+        try:
+            os.chdir(_REPO_ROOT)
+            specs, error = handler.resolve_test_group_specs("rust-server")
+            self.assertIsNone(error)
+            self.assertEqual(
+                specs,
+                [
+                    "registered/rust/test_run_rust_tests.py",
+                    "registered/core/test_srt_endpoint.py",
+                    "registered/vlm/test_rust_native_mm_e2e.py",
+                    "registered/vlm/test_rust_native_mm_mmmu.py",
+                ],
+            )
+
+            resolved = [
+                item
+                for test_spec in specs
+                for item in handler._resolve_test_spec(test_spec)
+            ]
+            self.assertTrue(all(item["error"] is None for item in resolved), resolved)
+            self.assertEqual(
+                [item["mode"] for item in resolved],
+                ["cpu", "cuda", "cuda", "cuda"],
+            )
+        finally:
+            os.chdir(previous_cwd)
+
+
+if __name__ == "__main__":
+    unittest.main()