From c70c7d72a8b1b0ef9ed8b37eede561ca070f3486 Mon Sep 17 00:00:00 2001 From: Cheng Wan <54331508+ch-wan@users.noreply.github.com> Date: Mon, 17 Aug 2026 16:18:19 -0700 Subject: [PATCH] config: the readback and the resolving view say what they are (#35027) --- .../sgl-router/src/workers/introspect.rs | 223 +++++++--- python/sglang/srt/entrypoints/engine.py | 25 +- python/sglang/srt/entrypoints/grpc_bridge.py | 8 +- python/sglang/srt/entrypoints/http_server.py | 23 +- python/sglang/srt/utils/common.py | 22 +- rust/sglang-server/src/api_server/common.rs | 17 +- rust/sglang-server/src/runtime/config.rs | 12 + .../steps/worker/local/discover_metadata.rs | 62 ++- .../test_effective_state_surfaces.py | 401 ++++++++++++++++++ .../unit/entrypoints/test_server_info.py | 71 +++- .../unit/test_global_config_read_ratchet.py | 6 + ...test_supplied_instance_exposure_ratchet.py | 2 - 12 files changed, 775 insertions(+), 97 deletions(-) create mode 100644 test/registered/unit/entrypoints/test_effective_state_surfaces.py diff --git a/experimental/sgl-router/src/workers/introspect.rs b/experimental/sgl-router/src/workers/introspect.rs index f6330bf69..1748b678a 100644 --- a/experimental/sgl-router/src/workers/introspect.rs +++ b/experimental/sgl-router/src/workers/introspect.rs @@ -1,23 +1,28 @@ // SPDX-FileCopyrightText: Copyright (c) 2026 The SGLang Authors // SPDX-License-Identifier: Apache-2.0 -//! Single-shot `/server_info` introspection for newly-discovered workers. +//! Introspection for newly-discovered workers. //! -//! Combines what used to be two separate round-trips (the worker -//! manager's `served_model_name` fetch and `KvEventIndex::add_worker`'s -//! `fetch_event_config`) into one HTTP request. The result is dispatched -//! by the manager: registry consumes `served_model_name`, the optional -//! `KvEventIndex` consumes the resolved `EventConfig`. +//! Two concurrent requests, because the worker answers two different +//! questions on two different endpoints: `/model_info` reports the identity +//! the worker currently serves under (a weight update moves it), while +//! `/server_info` reports its launch configuration — kv-event publisher and +//! disaggregation role. The result is dispatched by the manager: registry +//! consumes `served_model_name`, the optional `KvEventIndex` consumes the +//! resolved `EventConfig`. +//! +//! `served_model_name` is taken from `/model_info`, falling back to +//! `/server_info` for workers that predate the field there. //! //! # Failure semantics //! //! `fetch` is **infallible** — any error (network, non-2xx, JSON parse, -//! invalid worker URL) is logged at `warn!` and returns an empty -//! `ServerInfo` so the caller can register the worker with empty -//! `model_ids` and no kv-events attachment. Workers that need accuracy -//! around publisher availability use `kv_events::discovery::fetch_event_config` -//! directly (it returns `Result>`); the manager -//! intentionally doesn't. +//! invalid worker URL) is logged and yields `None` for whatever that request +//! was carrying, so the caller can register the worker with empty `model_ids` +//! and no kv-events attachment. The two requests fail independently. Workers +//! that need accuracy around publisher availability use +//! `kv_events::discovery::fetch_event_config` directly (it returns +//! `Result>`); the manager intentionally doesn't. use std::time::Duration; @@ -68,9 +73,9 @@ pub enum DisaggregationRole { Decode, } -/// Performs the single `/server_info` round-trip and projects the -/// response into both halves of `ServerInfo`. Cheap to clone — wraps a -/// `reqwest::Client` (which is internally `Arc`-backed). +/// Performs the two round-trips concurrently and projects the responses into +/// `ServerInfo`. Cheap to clone — wraps a `reqwest::Client` (which is +/// internally `Arc`-backed). #[derive(Clone)] pub struct WorkerIntrospector { client: reqwest::Client, @@ -95,9 +100,9 @@ impl WorkerIntrospector { Self { client } } - /// Fetch `/server_info` for the worker. Never returns an error: - /// any failure is logged at `warn!` and yields a default - /// `ServerInfo` with both halves `None`. Callers register the + /// Fetch `/model_info` and `/server_info` for the worker, concurrently. + /// Never returns an error: any failure is logged at `warn!` and yields + /// `None` for the fields that request carried. Callers register the /// worker with empty model IDs and no event subscription on the /// failure path; future re-discovery will retry. /// @@ -106,24 +111,31 @@ impl WorkerIntrospector { /// responses and JSON-parse errors short-circuit immediately — /// the worker answered authoritatively, retrying won't help. pub async fn fetch(&self, worker_url: &str) -> ServerInfo { - let server_info_url = format!("{}/server_info", worker_url.trim_end_matches('/')); - let parsed = match Self::fetch_with_retry(&self.client, &server_info_url, worker_url).await - { - Some(p) => p, - None => return ServerInfo::default(), - }; + let base = worker_url.trim_end_matches('/'); + let server_info_url = format!("{base}/server_info"); + let model_info_url = format!("{base}/model_info"); + let (parsed, model_info) = tokio::join!( + Self::fetch_with_retry::(&self.client, &server_info_url, worker_url), + Self::fetch_with_retry::(&self.client, &model_info_url, worker_url), + ); + // A worker that answers one endpoint and not the other still gets + // registered with whatever did answer. + let parsed = parsed.unwrap_or_default(); - let served_model_name = match parsed.served_model_name { - Some(name) if !name.is_empty() => Some(name), - Some(_) => { - warn!( - worker_url = %worker_url, - "introspect: /server_info has empty `served_model_name`; registering worker with empty model_ids" - ); - None - } - None => None, - }; + // `/model_info` is the effective identity; `/server_info` is the launch + // record, kept as the fallback for workers that predate the field + // there. An empty string is the same as absent on either. + let non_empty = |name: String| Some(name).filter(|name| !name.is_empty()); + let served_model_name = model_info + .and_then(|body| body.served_model_name) + .and_then(non_empty) + .or_else(|| parsed.served_model_name.and_then(non_empty)); + if served_model_name.is_none() { + warn!( + worker_url = %worker_url, + "introspect: no `served_model_name` on /model_info or /server_info; registering worker with empty model_ids" + ); + } // EAGLE-family speculative decoding ⇒ the worker hashes KV blocks over // token bigrams; the router must mirror that on the selection side. @@ -148,48 +160,53 @@ impl WorkerIntrospector { } } - /// Issue the `/server_info` GET with bounded retry on transient - /// errors. Returns `Some(body)` on success, `None` after exhausting - /// retries (the caller falls back to default `ServerInfo`). - async fn fetch_with_retry( + /// Issue one introspection GET with bounded retry on transient errors. + /// Returns `Some(body)` on success, `None` after exhausting retries or on + /// an authoritative answer (4xx, unparsable JSON). The caller decides what + /// a missing body costs — each endpoint carries different fields. + async fn fetch_with_retry( client: &reqwest::Client, - server_info_url: &str, + url: &str, worker_url: &str, - ) -> Option { + ) -> Option { let mut delay = FETCH_BACKOFF_BASE; for attempt in 1..=FETCH_MAX_ATTEMPTS { - match client.get(server_info_url).send().await { + match client.get(url).send().await { Err(e) => { warn!( worker_url = %worker_url, + url = %url, attempt, error = %e, - "introspect: /server_info request failed; will retry" + "introspect: request failed; will retry" ); } Ok(resp) if resp.status().is_server_error() => { warn!( worker_url = %worker_url, + url = %url, attempt, status = %resp.status(), - "introspect: /server_info returned 5xx; will retry" + "introspect: returned 5xx; will retry" ); } Ok(resp) if !resp.status().is_success() => { warn!( worker_url = %worker_url, + url = %url, status = %resp.status(), - "introspect: /server_info returned non-2xx; registering worker with empty model_ids" + "introspect: returned non-2xx; treating the endpoint as absent" ); return None; } - Ok(resp) => match resp.json::().await { + Ok(resp) => match resp.json::().await { Ok(body) => return Some(body), Err(e) => { warn!( worker_url = %worker_url, + url = %url, error = %e, - "introspect: /server_info JSON parse failed; registering worker with empty model_ids" + "introspect: JSON parse failed; treating the endpoint as absent" ); return None; } @@ -202,8 +219,9 @@ impl WorkerIntrospector { } warn!( worker_url = %worker_url, + url = %url, attempts = FETCH_MAX_ATTEMPTS, - "introspect: /server_info failed after retries; registering worker with empty model_ids" + "introspect: failed after retries; treating the endpoint as absent" ); None } @@ -306,11 +324,21 @@ pub(crate) fn resolve_event_config( } } +/// Projection of `/model_info` used by the introspector: the identity the +/// worker currently serves under, which a weight update moves. +#[derive(Debug, Default, Deserialize)] +struct ModelInfoBody { + #[serde(default)] + served_model_name: Option, +} + /// Projection of `/server_info` used by the introspector. Every field is /// `#[serde(default)]` so a worker that exposes only some of them still /// deserialises; downstream callers handle `None` as "absent". #[derive(Debug, Default, Deserialize)] struct ServerInfoBody { + /// The launch record's value, and the fallback for a worker whose + /// `/model_info` predates the field. #[serde(default)] served_model_name: Option, #[serde(default)] @@ -359,17 +387,36 @@ mod tests { use tokio::net::TcpListener; use tokio::sync::oneshot; + /// A worker that serves `/server_info` only: an SGLang that predates + /// `served_model_name` on `/model_info`. async fn spawn_fake_worker(body: Value) -> (String, oneshot::Sender<()>) { - let body = Arc::new(body); + spawn_fake_worker_with_model_info(body, None).await + } + + async fn spawn_fake_worker_with_model_info( + server_info: Value, + model_info: Option, + ) -> (String, oneshot::Sender<()>) { + let body = Arc::new(server_info); let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); let port = listener.local_addr().unwrap().port(); - let app = Router::new().route( + let mut app = Router::new().route( "/server_info", get(move || { let body = body.clone(); async move { Json((*body).clone()) } }), ); + if let Some(model_info) = model_info { + let model_info = Arc::new(model_info); + app = app.route( + "/model_info", + get(move || { + let model_info = model_info.clone(); + async move { Json((*model_info).clone()) } + }), + ); + } let (tx, rx) = oneshot::channel::<()>(); tokio::spawn(async move { let _ = axum::serve(listener, app) @@ -500,6 +547,82 @@ mod tests { ); } + #[tokio::test] + async fn fetch_prefers_model_info_over_server_info_for_the_name() { + let (url, _shutdown) = spawn_fake_worker_with_model_info( + json!({"served_model_name": "at-launch"}), + Some(json!({"served_model_name": "after-weight-update"})), + ) + .await; + let got = fast_introspector().fetch(&url).await; + assert_eq!( + got.served_model_name.as_deref(), + Some("after-weight-update") + ); + } + + #[tokio::test] + async fn fetch_falls_back_to_server_info_when_model_info_lacks_the_name() { + let (url, _shutdown) = spawn_fake_worker_with_model_info( + json!({"served_model_name": "at-launch"}), + Some(json!({"model_path": "/models/m"})), + ) + .await; + let got = fast_introspector().fetch(&url).await; + assert_eq!(got.served_model_name.as_deref(), Some("at-launch")); + } + + /// An empty name on either surface is the same as absent. + #[tokio::test] + async fn fetch_treats_an_empty_name_as_absent_on_both() { + let (url, _shutdown) = spawn_fake_worker_with_model_info( + json!({"served_model_name": "at-launch"}), + Some(json!({"served_model_name": ""})), + ) + .await; + assert_eq!( + fast_introspector() + .fetch(&url) + .await + .served_model_name + .as_deref(), + Some("at-launch"), + ); + + let (url, _shutdown) = + spawn_fake_worker_with_model_info(json!({"served_model_name": ""}), None).await; + assert!(fast_introspector() + .fetch(&url) + .await + .served_model_name + .is_none()); + } + + #[tokio::test] + async fn fetch_keeps_the_name_when_server_info_is_absent() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = listener.local_addr().unwrap().port(); + let app = Router::new().route( + "/model_info", + get(|| async { Json(json!({"served_model_name": "m"})) }), + ); + let (tx, rx) = oneshot::channel::<()>(); + tokio::spawn(async move { + let _ = axum::serve(listener, app) + .with_graceful_shutdown(async move { + let _ = rx.await; + }) + .await; + }); + let _shutdown = tx; + let got = fast_introspector() + .fetch(&format!("http://127.0.0.1:{port}")) + .await; + assert_eq!(got.served_model_name.as_deref(), Some("m")); + assert!(got.event_config.is_none()); + assert!(got.disaggregation_role.is_none()); + } + #[tokio::test] async fn fetch_only_served_model_name_when_kv_events_absent() { let (url, _shutdown) = spawn_fake_worker(json!({"served_model_name": "m"})).await; diff --git a/python/sglang/srt/entrypoints/engine.py b/python/sglang/srt/entrypoints/engine.py index 43bb046ee..ca2c429fc 100644 --- a/python/sglang/srt/entrypoints/engine.py +++ b/python/sglang/srt/entrypoints/engine.py @@ -1364,9 +1364,7 @@ class Engine(EngineScoreMixin, EngineBase): ) return msgspec_to_builtins( { - **self.tokenizer_manager.resolved_config_dict( - dataclasses.asdict(self.tokenizer_manager.server_args) - ), + **dataclasses.asdict(self.tokenizer_manager.server_args), **self._scheduler_init_result.scheduler_infos[0], "startup_time": self.tokenizer_manager.startup_time, "internal_states": internal_states, @@ -1374,6 +1372,27 @@ class Engine(EngineScoreMixin, EngineBase): } ) + def get_model_info(self): + """What this engine is serving right now. + + `get_server_info` answers with the record: the launch configuration, + parsers included -- `auto` resolves into the record before the config + is published. This surface adds what the control plane changed after + publication: the model a weight update swapped in, its load format, an + operator-set weight version. The HTTP and gRPC model-info endpoints + answer with the same fields. + """ + tm = self.tokenizer_manager + return { + "model_path": tm.model_path, + "served_model_name": tm.served_model_name, + "is_generation": tm.is_generation, + "weight_version": tm.config_value("weight_version"), + "load_format": tm.config_value("load_format"), + "reasoning_parser": tm.config_value("reasoning_parser"), + "tool_call_parser": tm.config_value("tool_call_parser"), + } + def init_weights_update_group( self, master_address: str, diff --git a/python/sglang/srt/entrypoints/grpc_bridge.py b/python/sglang/srt/entrypoints/grpc_bridge.py index 176503a50..1139756fc 100644 --- a/python/sglang/srt/entrypoints/grpc_bridge.py +++ b/python/sglang/srt/entrypoints/grpc_bridge.py @@ -397,9 +397,13 @@ class RuntimeHandle: model_config = self.tokenizer_manager.model_config result = { "model_path": self.tokenizer_manager.model_path, + "served_model_name": self.tokenizer_manager.served_model_name, "tokenizer_path": self.tokenizer_manager.server_args.tokenizer_path, "is_generation": self.tokenizer_manager.is_generation, "weight_version": self.tokenizer_manager.config_value("weight_version"), + "load_format": self.tokenizer_manager.config_value("load_format"), + "reasoning_parser": self.tokenizer_manager.config_value("reasoning_parser"), + "tool_call_parser": self.tokenizer_manager.config_value("tool_call_parser"), "model_type": getattr(model_config.hf_config, "model_type", None), "architectures": getattr(model_config.hf_config, "architectures", None), } @@ -413,9 +417,7 @@ class RuntimeHandle: return json.dumps(result, default=str) def get_server_info(self) -> str: - result: Dict[str, Any] = self.tokenizer_manager.resolved_config_dict( - dataclasses.asdict(self.tokenizer_manager.server_args) - ) + result: Dict[str, Any] = dataclasses.asdict(self.tokenizer_manager.server_args) result.update(self.scheduler_info) return json.dumps(msgspec_to_builtins(result), default=str) diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index 8e8ffee20..fe9697a08 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -741,12 +741,22 @@ async def model_info(): model_config = _global_state.tokenizer_manager.model_config result = { "model_path": _global_state.tokenizer_manager.model_path, + # Manager-owned, and moved by a weight update alongside `model_path`: + # this is where a client reads the identity the server answers under. + "served_model_name": _global_state.tokenizer_manager.served_model_name, "tokenizer_path": _global_state.tokenizer_manager.server_args.tokenizer_path, "is_generation": _global_state.tokenizer_manager.is_generation, "preferred_sampling_params": _global_state.tokenizer_manager.server_args.preferred_sampling_params, "weight_version": _global_state.tokenizer_manager.config_value( "weight_version" ), + "load_format": _global_state.tokenizer_manager.config_value("load_format"), + "reasoning_parser": _global_state.tokenizer_manager.config_value( + "reasoning_parser" + ), + "tool_call_parser": _global_state.tokenizer_manager.config_value( + "tool_call_parser" + ), "has_image_understanding": model_config.is_image_understandable_model, "has_audio_understanding": model_config.is_audio_understandable_model, "model_type": getattr(model_config.hf_config, "model_type", None), @@ -785,7 +795,14 @@ async def get_server_info(): @app.get("/server_info") async def server_info(): - """Get the server information.""" + """The startup configuration, plus live scheduler state. + + The `ServerArgs` fields here are the record: what the launcher was given, + with resolution written back into it. Fields the control plane changes + after publication -- the model a weight update swapped in, its load format, + an operator-set weight version -- are reported by `/model_info`, and the + HiCache mirror by `GET /hicache/storage-backend`. + """ # Returns internal states per DP. internal_states: List[Dict[Any, Any]] = ( await _global_state.tokenizer_manager.get_internal_state() @@ -796,9 +813,7 @@ async def server_info(): # server_args.model_config is not serializable but should be excluded by asdict. return msgspec_to_builtins( { - **_global_state.tokenizer_manager.resolved_config_dict( - dataclasses.asdict(server_args) - ), + **dataclasses.asdict(server_args), **_global_state.scheduler_info, "startup_time": _global_state.tokenizer_manager.startup_time, "internal_states": internal_states, diff --git a/python/sglang/srt/utils/common.py b/python/sglang/srt/utils/common.py index 6438b78e8..053f2cdb9 100644 --- a/python/sglang/srt/utils/common.py +++ b/python/sglang/srt/utils/common.py @@ -98,7 +98,11 @@ from typing_extensions import Literal from sglang.srt.environ import envs from sglang.srt.observability.func_timer import enable_func_timer from sglang.srt.platforms import current_platform -from sglang.srt.runtime_context import get_parallel +from sglang.srt.runtime_context import ( + configured_tp_size, + get_exec, + get_parallel, +) from sglang.srt.utils.video_decoder import _BACKEND, VideoDecoderWrapper if TYPE_CHECKING: @@ -3578,10 +3582,12 @@ def bind_or_assign(target, source): # TODO(hebiao064): Accelerate FA3 Spec Decode with topk > 1. # TODO(hebiao064): Improve the acc rate for FA3 Spec Decode with topk == 1 and page_size > 1. -def is_no_spec_infer_or_topk_one(server_args): - return server_args.speculative_eagle_topk is None or ( - server_args.speculative_eagle_topk == 1 - and (server_args.page_size == 1 or server_args.page_size is None) +def is_no_spec_infer_or_topk_one(cfg): + """``cfg`` is a resolving config view, not the published record: the + resolution pipeline is the only caller, and it asks mid-resolution.""" + return cfg.speculative_eagle_topk is None or ( + cfg.speculative_eagle_topk == 1 + and (cfg.page_size == 1 or cfg.page_size is None) ) @@ -3752,7 +3758,7 @@ def require_mlp_tp_gather(server_args: ServerArgs): else: return ( get_parallel().moe_dense_tp_size - > server_args.tp_size // get_parallel().dp_size + > configured_tp_size() // get_parallel().dp_size ) else: return False @@ -3778,7 +3784,7 @@ def require_attn_tp_gather(server_args: ServerArgs): or get_parallel().moe_dense_tp_size is not None ): if get_parallel().enable_dp_attention: - return get_parallel().dp_size < server_args.tp_size + return get_parallel().dp_size < configured_tp_size() else: return True else: @@ -3797,7 +3803,7 @@ def require_mlp_sync(server_args: ServerArgs): def get_cuda_graph_batch_size_alignment(server_args: ServerArgs) -> int: alignment = 1 - if server_args.enable_two_batch_overlap: + if get_exec().overlap.enable_two_batch_overlap: alignment *= 2 if require_gathered_buffer(server_args): alignment *= get_parallel().attn_tp_size diff --git a/rust/sglang-server/src/api_server/common.rs b/rust/sglang-server/src/api_server/common.rs index 0b079661a..e7f3025b5 100644 --- a/rust/sglang-server/src/api_server/common.rs +++ b/rust/sglang-server/src/api_server/common.rs @@ -70,10 +70,18 @@ async fn await_control_result( /// `GET /get_model_info` (+ `/model_info` alias) — static model metadata from /// `server_args` (no scheduler round-trip); `is_generation` always true. +/// +/// Under `SGLANG_RUST_SERVER=1` this is the only `/model_info` a client can +/// reach — `launch_server` never mounts the Python app — so it answers the same +/// keys. It answers them from the launch blob, which is the whole of this +/// server's config knowledge: `server_args` is parsed once at boot and held +/// behind an `Arc`, and no route mounted here changes weights or parsers, so +/// the launch values are also the current ones. async fn model_info(State(state): State) -> Response { let sa = &state.server_args; let body = serde_json::json!({ "model_path": sa.model_path, + "served_model_name": sa.served_model_name, "tokenizer_path": sa.tokenizer_path, "is_generation": true, // Python's `TokenizerManager` merges this into every request @@ -81,7 +89,14 @@ async fn model_info(State(state): State) -> Response { // `RustServer.launch` REFUSES to start when it is set. It can therefore // only be null here — echoing it keeps the field's shape. "preferred_sampling_params": sa.preferred_sampling_params, - "weight_version": serde_json::Value::Null, + // Python answers this through `config_value`, so a control-plane write + // moves it there; here it is the launch value. + "weight_version": sa.weight_version, + "load_format": sa.load_format, + // `auto` never reaches the blob: `resolve_auto_parsers` writes the + // selected parser into `server_args` before the scheduler forks. + "reasoning_parser": sa.reasoning_parser, + "tool_call_parser": sa.tool_call_parser, }); ( StatusCode::OK, diff --git a/rust/sglang-server/src/runtime/config.rs b/rust/sglang-server/src/runtime/config.rs index b3c91d9ca..b12a1294f 100644 --- a/rust/sglang-server/src/runtime/config.rs +++ b/rust/sglang-server/src/runtime/config.rs @@ -72,6 +72,18 @@ pub struct ServerArgs { /// HF revision, used only when `tokenizer_path` is a repo id. `None` → main. #[serde(default)] pub revision: Option, + /// Weight format selected by `--load-format`, reported by `/get_model_info`. + /// The blob carries the post-`__post_init__` value (`auto` is already + /// narrowed to `gguf` / `mistral` / `runai_streamer` / `remote` where the + /// checkpoint demands it). Not consumed for loading -- the scheduler owns + /// that; `None` only when the blob omits the key. + #[serde(default)] + pub load_format: Option, + /// Operator-supplied weight version, reported by `/model_info`. Defaults to + /// `"default"` on the Python side, so it is present in every blob; `None` + /// only when the blob omits the key. + #[serde(default)] + pub weight_version: Option, /// HTTP bind address (see [`Self::bind`]). #[serde(default = "default_host")] pub host: String, diff --git a/sgl-model-gateway/src/core/steps/worker/local/discover_metadata.rs b/sgl-model-gateway/src/core/steps/worker/local/discover_metadata.rs index 026772688..0125acb17 100644 --- a/sgl-model-gateway/src/core/steps/worker/local/discover_metadata.rs +++ b/sgl-model-gateway/src/core/steps/worker/local/discover_metadata.rs @@ -47,6 +47,10 @@ pub struct ServerInfo { #[derive(Debug, Clone, Deserialize, Serialize)] pub struct ModelInfo { pub model_path: Option, + /// The identity the worker currently serves under. A weight update moves + /// this (and `model_path`) on the worker's manager, so it is answered here + /// rather than by `/server_info`, which reports the launch record. + pub served_model_name: Option, pub tokenizer_path: Option, pub is_generation: Option, pub model_type: Option, @@ -277,18 +281,50 @@ impl StepExecutor for DiscoverMetadataStep { ConnectionMode::Http => { let mut labels = HashMap::new(); - // Fetch from /server_info for server-related metadata - if let Ok(server_info) = - get_server_info(&config.url, config.api_key.as_deref()).await - { - if let Some(model_path) = server_info.model_path.filter(|s| !s.is_empty()) { - labels.insert("model_path".to_string(), model_path); - } - if let Some(served_model_name) = - server_info.served_model_name.filter(|s| !s.is_empty()) - { - labels.insert("served_model_name".to_string(), served_model_name); + // /server_info reports the launch configuration; /model_info + // reports the model the worker is serving now. Both are read + // here, and a failure of one does not lose the other. + let server_info = get_server_info(&config.url, config.api_key.as_deref()) + .await + .ok(); + let model_info = get_model_info(&config.url, config.api_key.as_deref()) + .await + .ok(); + + // Identity comes from /model_info, which a weight update moves; + // /server_info answers the launch record and is the fallback for + // workers that predate the fields there. + let present = |value: Option<&String>| value.filter(|s| !s.is_empty()).cloned(); + let identity = [ + ( + "model_path", + present(model_info.as_ref().and_then(|m| m.model_path.as_ref())).or_else( + || present(server_info.as_ref().and_then(|s| s.model_path.as_ref())), + ), + ), + ( + "served_model_name", + present( + model_info + .as_ref() + .and_then(|m| m.served_model_name.as_ref()), + ) + .or_else(|| { + present( + server_info + .as_ref() + .and_then(|s| s.served_model_name.as_ref()), + ) + }), + ), + ]; + for (key, value) in identity { + if let Some(value) = value { + labels.insert(key.to_string(), value); } + } + + if let Some(server_info) = server_info { if let Some(tp_size) = server_info.tp_size { labels.insert("tp_size".to_string(), tp_size.to_string()); } @@ -303,9 +339,7 @@ impl StepExecutor for DiscoverMetadataStep { } } - // Fetch from /model_info for model-related metadata - if let Ok(model_info) = get_model_info(&config.url, config.api_key.as_deref()).await - { + if let Some(model_info) = model_info { if let Some(tokenizer_path) = model_info.tokenizer_path.filter(|s| !s.is_empty()) { diff --git a/test/registered/unit/entrypoints/test_effective_state_surfaces.py b/test/registered/unit/entrypoints/test_effective_state_surfaces.py new file mode 100644 index 000000000..c1434d8cc --- /dev/null +++ b/test/registered/unit/entrypoints/test_effective_state_surfaces.py @@ -0,0 +1,401 @@ +"""Every serving surface can answer what is running, not only what was asked. + +`/server_info` and its gRPC and in-process twins report the startup record, +parsers included -- the launcher resolves `auto` into the record before +publishing. What changes after publication -- the model a weight update +swapped in, its load format, an operator-set weight version -- is reported by +the model-info surface, and there is one per entry point: HTTP, gRPC and +`Engine`. Adding a field to one and forgetting the others leaves that entry +point's users with no way to see it, which no test notices because each +surface passes its own tests. + +The required set has two halves. The derived half comes from the control-plane +writers: whatever a process writes with `override` after publication is exactly +what can differ from the record, and both the keyword and the `**`-expansion +shapes resolve statically here. The second half is a policy, not a derivation -- +what any one surface reports effectively, all of them owe their users -- so a +field every surface drops at once leaves the set with it. Each surface must both +carry the key and take its value from the effective config: reading +`server_args.` under the right key reports the startup value with a +straight face. +""" + +import ast +import inspect +import pathlib +import re +import unittest + +import sglang +from sglang.srt.entrypoints.engine import Engine +from sglang.test.ci.ci_register import register_cpu_ci +from sglang.test.test_utils import CustomTestCase + +register_cpu_ci(est_time=3, suite="base-a-test-cpu") + +_PACKAGE_ROOT = pathlib.Path(sglang.__file__).resolve().parent +_RUST_MODEL_INFO = ( + pathlib.Path(__file__).resolve().parents[4] + / "rust/sglang-server/src/api_server/common.rs" +) + +# The tokenizer process is the one whose control-plane writes a served request +# can observe; a field it overrides there is a post-launch fact. +_TOKENIZER_WRITERS = ( + "srt/managers/tokenizer_manager.py", + "srt/managers/tokenizer_control_mixin.py", + "srt/entrypoints/http_server.py", + "srt/entrypoints/engine.py", +) + +# The model path and the served name stay manager attributes: a weight update +# moves them on the manager rather than through `override`, so the writer +# derivation cannot see them. They are subtracted from the derived half and +# asserted directly instead -- an exemption whose premise ("every surface +# reports them") is checked, not assumed. It was not true when it was written: +# only `Engine` carried `served_model_name`, and the router had to read the +# launch record off `/server_info` to learn a name a weight update had moved. +_MANAGER_ATTRIBUTES = {"model_path", "served_model_name"} + + +def _hicache_status_fields() -> set: + """The fields `GET /hicache/storage-backend` answers with. + + The HiCache mirror is written post-publish and reported by its own + endpoint, so the model-info surfaces do not owe it. Taking the set from + that handler is what keeps the exemption honest: a field it stops + reporting falls back to them. + """ + tree = ast.parse((_PACKAGE_ROOT / "srt/entrypoints/http_server.py").read_text()) + for fn in ast.walk(tree): + if ( + not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) + or fn.name != "hicache_storage_backend_status" + ): + continue + fields = { + elt.value + for comp in ast.walk(fn) + if isinstance(comp, ast.DictComp) + for gen in comp.generators + for elt in getattr(gen.iter, "elts", []) + if isinstance(elt, ast.Constant) and isinstance(elt.value, str) + } + assert fields, "the HiCache status handler names no field" + return fields + raise AssertionError( + "no `hicache_storage_backend_status` handler: the HiCache fields have " + "no endpoint of their own and fall to the model-info surfaces" + ) + + +def _expanded_write_keys(rel: str, tree: ast.AST, call: ast.Call, kw: ast.keyword): + """The field names behind a `**` at a control-plane writer call. + + Resolves a dict literal -- constant keys, or a key bound by an enclosing + literal `for` -- and a name bound to a dict literal in the enclosing + function, including constant-subscript stores onto it. A `**` that + forwards its own function's `**kwargs` names no field: its callers do. + Anything else raises, because a skipped expansion shrinks the required + set instead of failing. + """ + enclosing = None + for fn in ast.walk(tree): + if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)) and ( + fn.lineno <= call.lineno <= (fn.end_lineno or fn.lineno) + ): + if enclosing is None or fn.lineno > enclosing.lineno: + enclosing = fn + + def loop_bound(name: str) -> set: + """The values a `for name, ... in ()` around the call binds.""" + values = set() + for node in ast.walk(tree): + if not isinstance(node, ast.For): + continue + target = node.target + names = ( + [target] + if isinstance(target, ast.Name) + else list(getattr(target, "elts", [])) + ) + if not names or not isinstance(names[0], ast.Name) or names[0].id != name: + continue + if not (node.lineno <= call.lineno <= (node.end_lineno or node.lineno)): + continue + for item in getattr(node.iter, "elts", []): + first = ( + item.elts[0] if isinstance(item, ast.Tuple) and item.elts else item + ) + if isinstance(first, ast.Constant) and isinstance(first.value, str): + values.add(first.value) + return values + + def dict_keys(node: ast.Dict) -> set: + keys = set() + for key in node.keys: + if isinstance(key, ast.Constant): + keys.add(key.value) + continue + assert isinstance( + key, ast.Name + ), f"non-literal dict key in a writer expansion at {rel}:{call.lineno}" + bound = loop_bound(key.id) + assert bound, ( + f"dict key {key.id!r} at {rel}:{call.lineno} is not bound by a " + "literal loop; extend the resolver" + ) + keys |= bound + return keys + + if isinstance(kw.value, ast.Dict): + return dict_keys(kw.value) + assert isinstance( + kw.value, ast.Name + ), f"unresolvable writer expansion at {rel}:{call.lineno}" + assert ( + enclosing is not None + ), f"writer expansion outside any function at {rel}:{call.lineno}" + name = kw.value.id + if enclosing.args.kwarg is not None and enclosing.args.kwarg.arg == name: + return set() + keys = set() + found = False + for node in ast.walk(enclosing): + if not (isinstance(node, ast.Assign) and len(node.targets) == 1): + continue + target = node.targets[0] + if isinstance(target, ast.Name) and target.id == name: + assert isinstance(node.value, ast.Dict), ( + f"writer expansion {name!r} at {rel}:{call.lineno} is assigned " + "something other than a dict literal; extend the resolver" + ) + found = True + keys |= dict_keys(node.value) + elif ( + isinstance(target, ast.Subscript) + and isinstance(target.value, ast.Name) + and target.value.id == name + and isinstance(target.slice, ast.Constant) + ): + keys.add(target.slice.value) + assert found, ( + f"writer expansion {name!r} at {rel}:{call.lineno} has no dict-literal " + "assignment in its function; extend the resolver" + ) + return keys + + +def _overridden_fields() -> set: + """Fields the tokenizer process writes after publication.""" + fields = set() + for rel in _TOKENIZER_WRITERS: + tree = ast.parse((_PACKAGE_ROOT / rel).read_text()) + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + name = ( + node.func.attr + if isinstance(node.func, ast.Attribute) + else getattr(node.func, "id", "") + ) + if name not in ("override", "record_config_updates"): + continue + for kw in node.keywords: + if kw.arg == "source": + # The provenance label, not a config field. + continue + if kw.arg: + fields.add(kw.arg) + else: + fields |= _expanded_write_keys(rel, tree, node, kw) + # Only the mirror's own fields earn the endpoint exemption: adding an + # unrelated field to that handler must not buy it a pass here. + hicache = {f for f in _hicache_status_fields() if f.startswith("hicache_")} + return fields - _MANAGER_ATTRIBUTES - hicache + + +def _effective_reads_in(source: str, func_name: str) -> set: + """Fields the function reports *and* reads through the effective config. + + A key whose value comes off the `ServerArgs` record does not count: that is + the startup value under a name that promises the running one. + """ + tree = ast.parse(source) + for fn in ast.walk(tree): + if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if fn.name != func_name: + continue + reported = set() + for node in ast.walk(fn): + if not isinstance(node, ast.Dict): + continue + for key, value in zip(node.keys, node.values): + if not (isinstance(key, ast.Constant) and isinstance(key.value, str)): + continue + for inner in ast.walk(value): + if ( + isinstance(inner, ast.Call) + and isinstance(inner.func, ast.Attribute) + and inner.func.attr in ("config_value", "config_leaf") + and inner.args + and isinstance(inner.args[0], ast.Constant) + and inner.args[0].value == key.value + ): + reported.add(key.value) + return reported + return set() + + +def _rust_model_info_keys() -> set: + """The keys the rust server's `/model_info` handler answers with. + + Scanned as text, from the handler's signature to the next item in the + file, with each line cut at its first `//`. A key is always left of its + value, so cutting inside a string can only drop keys, never invent one. + The signature must appear exactly once: a rust handler that moved or was + renamed would otherwise contribute an empty set and let the parity check + pass on nothing. + """ + source = _RUST_MODEL_INFO.read_text() + marker = "async fn model_info(" + found = source.count(marker) + assert found == 1, ( + f"{_RUST_MODEL_INFO.name} declares `{marker}` {found} times; the rust " + "/model_info surface is the one users reach under SGLANG_RUST_SERVER=1 " + "and is no longer being read" + ) + body = source[source.index(marker) + len(marker) :] + following = re.search(r"^(?:pub(?:\([^)]*\))?\s+)?(?:async\s+)?fn\s", body, re.M) + if following is not None: + body = body[: following.start()] + code = "\n".join(line.split("//")[0] for line in body.splitlines()) + keys = set(re.findall(r'"([A-Za-z_][A-Za-z0-9_]*)"\s*:', code)) + assert keys, "the rust /model_info handler names no field" + return keys + + +def _reported_keys_in(source: str, func_name: str) -> set: + """Every string key the function's response dicts carry, whatever the value. + + The manager-owned attributes are read off the manager, not through + `config_value`, so `_effective_reads_in` does not see them; this is how they + are checked. + """ + tree = ast.parse(source) + for fn in ast.walk(tree): + if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)): + continue + if fn.name != func_name: + continue + return { + key.value + for node in ast.walk(fn) + if isinstance(node, ast.Dict) + for key in node.keys + if isinstance(key, ast.Constant) and isinstance(key.value, str) + } + return set() + + +def _model_info_sources() -> dict: + """`(source, function name)` per Python model-info surface.""" + return { + "http /model_info": ( + (_PACKAGE_ROOT / "srt/entrypoints/http_server.py").read_text(), + "model_info", + ), + "grpc get_model_info": ( + (_PACKAGE_ROOT / "srt/entrypoints/grpc_bridge.py").read_text(), + "get_model_info", + ), + "Engine.get_model_info": ( + inspect.getsource(Engine.get_model_info).lstrip(), + "get_model_info", + ), + } + + +def _model_info_surfaces() -> dict: + """Each Python entry point's model-info surface, by what it reports + effectively.""" + return { + name: _effective_reads_in(source, func_name) + for name, (source, func_name) in _model_info_sources().items() + } + + +class TestEffectiveStateSurfaces(CustomTestCase): + def test_each_entry_point_reports_the_post_launch_facts(self): + surfaces = _model_info_surfaces() + written = _overridden_fields() + # Both writer shapes are in reach of the scan: `load_format` is a + # literal keyword, the parsers arrive as `**{attr: ...}` under a key + # the enclosing loop binds. + self.assertLessEqual( + {"load_format", "reasoning_parser", "tool_call_parser"}, + written, + "the derivation stopped finding the control-plane writers", + ) + # What any surface reports effectively, all of them owe their users; + # what the control plane overrides, every surface owes regardless. + required = set().union(*surfaces.values()) | written + missing = { + name: sorted(required - reported) + for name, reported in surfaces.items() + if required - reported + } + self.assertEqual( + missing, + {}, + "a serving surface cannot report what it is running: " f"{missing}", + ) + + def test_every_surface_reports_the_manager_owned_identity(self): + """The identity a weight update moves is answered where it is read. + + `model_path` and `served_model_name` live on the tokenizer manager, so + the writer derivation cannot reach them and they are subtracted from the + required set. This is the assertion that pays for that subtraction. A + surface that drops one sends its clients back to the launch record -- + which is what the rust router had to read, under a name that had since + moved. + """ + surfaces = { + name: _reported_keys_in(source, func_name) + for name, (source, func_name) in _model_info_sources().items() + } + surfaces["rust /model_info"] = _rust_model_info_keys() + missing = { + name: sorted(_MANAGER_ATTRIBUTES - reported) + for name, reported in surfaces.items() + if _MANAGER_ATTRIBUTES - reported + } + self.assertEqual( + missing, + {}, + f"a model-info surface does not say which model it serves: {missing}", + ) + + def test_the_rust_model_info_answers_the_same_keys(self): + """`SGLANG_RUST_SERVER=1` swaps the whole HTTP server, not one handler. + + The keys are owed there too, or the endpoint's contract depends on + which server the operator launched. The values are the launch record: + that process parses `server_args` once and mounts no route that can + change weights or parsers, so this is a key-set check and the handler + states which it reports. + """ + required = set().union(*_model_info_surfaces().values()) | _overridden_fields() + missing = sorted(required - _rust_model_info_keys()) + self.assertEqual( + missing, + [], + "the rust /model_info answers a different contract than the Python " + f"one it replaces: {missing}", + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/registered/unit/entrypoints/test_server_info.py b/test/registered/unit/entrypoints/test_server_info.py index 2cb019634..fe6d9e0f8 100644 --- a/test/registered/unit/entrypoints/test_server_info.py +++ b/test/registered/unit/entrypoints/test_server_info.py @@ -29,6 +29,7 @@ from types import SimpleNamespace from sglang.srt.entrypoints import http_server from sglang.srt.lora.lora_registry import LoRARef from sglang.srt.managers.tokenizer_manager import TokenizerManager +from sglang.srt.runtime_context import publish, reset_context from sglang.srt.server_args import ServerArgs from sglang.test.ci.ci_register import register_cpu_ci from sglang.test.test_utils import CustomTestCase @@ -36,6 +37,25 @@ from sglang.test.test_utils import CustomTestCase register_cpu_ci(est_time=5, suite="base-a-test-cpu") +def _stub_tokenizer_manager( + server_args: ServerArgs, get_internal_state=None +) -> TokenizerManager: + """A manager carrying the state `/server_info` and its writers read. + + `__new__` skips `__init__`, which would open the ZMQ sockets and start + the handle loop; `_config_updates` is the log `record_config_updates` + appends to. + """ + tokenizer_manager = TokenizerManager.__new__(TokenizerManager) + tokenizer_manager.server_args = server_args + tokenizer_manager.model_path = server_args.model_path + tokenizer_manager.served_model_name = server_args.served_model_name + tokenizer_manager.startup_time = None + tokenizer_manager._config_updates = [] + tokenizer_manager.get_internal_state = get_internal_state + return tokenizer_manager + + def _call_server_info_with( server_args: ServerArgs, internal_states: list[dict] | None = None, @@ -48,31 +68,37 @@ def _call_server_info_with( `SimpleNamespace` stub via `set_global_state` and awaiting the coroutine directly is enough to exercise the handler logic without booting a model server. + + `config_updates` are applied the way production applies them -- through + `record_config_updates`, in a process that has published -- rather than + planted on the stub. The endpoint answers the record and does not read that + log, so what the real writer buys is that an assertion on the record is not + satisfied for free by an update that never landed. """ async def _fake_internal_state(): return internal_states or [{"max_req_input_len": 1024}] - tokenizer_manager = TokenizerManager.__new__(TokenizerManager) - tokenizer_manager.server_args = server_args - tokenizer_manager.model_path = server_args.model_path - tokenizer_manager.served_model_name = server_args.served_model_name - tokenizer_manager.startup_time = None - tokenizer_manager._config_updates = ( - [("test", dict(config_updates))] if config_updates else [] - ) - tokenizer_manager.get_internal_state = _fake_internal_state + tokenizer_manager = _stub_tokenizer_manager(server_args, _fake_internal_state) stub_state = SimpleNamespace( tokenizer_manager=tokenizer_manager, scheduler_info={"max_req_input_len": 1024}, ) prior_state = http_server.get_global_state() http_server.set_global_state(stub_state) + published = False + if config_updates: + # The writer runs in a published process, as it does in production. + publish(server_args, role="tokenizer") + published = True + tokenizer_manager.record_config_updates("test", **config_updates) try: return asyncio.run(http_server.server_info()) finally: # Restore so a later test in the same process isn't surprised. http_server._global_state = prior_state + if published: + reset_context() class TestServerInfoKvEventsField(CustomTestCase): @@ -243,16 +269,37 @@ class TestServerInfoKvEventsField(CustomTestCase): class TestServerInfoControlPlaneUpdates(CustomTestCase): - """Runtime control-plane updates live on the manager, not on ServerArgs.""" + """/server_info answers what was asked for, not what is in effect.""" - def test_recorded_updates_win_over_the_startup_config(self): + def test_the_readback_reports_the_record_not_the_control_plane(self): + # A runtime weight-version change is reported by /model_info and its + # gRPC / Engine twins; this endpoint reports what the operator supplied. server_args = ServerArgs(model_path="dummy", weight_version="v1") payload = _call_server_info_with( server_args, config_updates={"weight_version": "v2"} ) - self.assertEqual(payload["weight_version"], "v2") + self.assertEqual(payload["weight_version"], "v1") self.assertEqual(server_args.weight_version, "v1") + def test_the_recorded_update_reaches_the_readback_overlay(self): + """The update the case above records is one `/server_info` could see. + + `resolved_config_dict` is the overlay the endpoint dropped; an update + that never reached it would satisfy the assertion above for free. + """ + server_args = ServerArgs(model_path="dummy", weight_version="v1") + tokenizer_manager = _stub_tokenizer_manager(server_args) + publish(server_args, role="tokenizer") + try: + tokenizer_manager.record_config_updates("test", weight_version="v2") + self.assertEqual(tokenizer_manager.config_value("weight_version"), "v2") + overlaid = tokenizer_manager.resolved_config_dict( + dataclasses.asdict(server_args) + ) + self.assertEqual(overlaid["weight_version"], "v2") + finally: + reset_context() + class TestServerInfoExistingFieldsPreserved(CustomTestCase): """Regression guard: the new `kv_events` field is additive — none of diff --git a/test/registered/unit/test_global_config_read_ratchet.py b/test/registered/unit/test_global_config_read_ratchet.py index fdbeb8677..3d447b66e 100644 --- a/test/registered/unit/test_global_config_read_ratchet.py +++ b/test/registered/unit/test_global_config_read_ratchet.py @@ -112,6 +112,12 @@ _CONFIGURED_SIZE_CALL_SITES = { "the encode server's launch entry sizes its workers before it has " "spawned any of them" ), + ("srt/utils/common.py", "configured_tp_size"): ( + "the require_*_tp_gather predicates compared the configured tp_size " + "when they read the record; the live property answers a different " + "question wherever the groups alias, so the configured accessor is the " + "mechanical substitution and the live one would be a semantic change" + ), ("srt/model_loader/loader.py", "configured_moe_dp_size"): ( "the same dict already carries the live moe_dp_size under 'dp'; this entry " "is the configured intent" diff --git a/test/registered/unit/test_supplied_instance_exposure_ratchet.py b/test/registered/unit/test_supplied_instance_exposure_ratchet.py index 0a3b39de7..a929a6c2f 100644 --- a/test/registered/unit/test_supplied_instance_exposure_ratchet.py +++ b/test/registered/unit/test_supplied_instance_exposure_ratchet.py @@ -349,8 +349,6 @@ _EXPOSED = { ("speculative/standalone_worker_v2.py", "speculative_eagle_topk"), ("speculative/standalone_worker_v2.py", "speculative_num_draft_tokens"), ("speculative/standalone_worker_v2.py", "speculative_num_steps"), - ("utils/common.py", "page_size"), - ("utils/common.py", "speculative_eagle_topk"), ("utils/common.py", "speculative_num_draft_tokens"), ("utils/common.py", "speculative_num_steps"), ("utils/cuda_vmm_transport_utils.py", "mm_feature_transport"),