[Feat] rust-server: Report weight versions and media token counts in generate meta_info (#46205)

Co-authored-by: metamergebot <metamergebot@users.noreply.github.com>
Co-authored-by: shodoco <shodoco@users.noreply.github.com>

proto/sglang/api/v1/generate.proto

@@ -205,6 +205,10 @@ message GenerateMetaInfo {
   optional string weight_version = 16;
   // Omitted until terminal output; reported spans are always nonempty.
   repeated WeightVersionSpan weight_versions = 17;
+  // Media prompt tokens per modality, on every frame; omitted when zero.
+  optional uint32 image_tokens = 18;
+  optional uint32 audio_tokens = 19;
+  optional uint32 video_tokens = 20;
 }
 
 message WeightVersionSpan {

python/sglang/api/v1/api_types.py

@@ -374,6 +374,9 @@ class GenerateMetaInfo(msgspec.Struct, kw_only=True, frozen=True):
     output_token_logprobs_length: Optional[int] = None
     weight_version: Optional[str] = None
     weight_versions: List[WeightVersionSpan] = []
+    image_tokens: Optional[int] = None
+    audio_tokens: Optional[int] = None
+    video_tokens: Optional[int] = None
 
     @classmethod
     def from_json_value(cls, value: Any) -> "GenerateMetaInfo":
@@ -1776,6 +1779,12 @@ def decode_GenerateMetaInfo(value: Any) -> GenerateMetaInfo:
             kw["weight_version"] = None if x is None else _expect_str(x, expected="a string")
         elif key == "weight_versions":
             kw["weight_versions"] = [] if x is None else [decode_WeightVersionSpan(e) for e in _expect_seq(x, expected="a sequence")]
+        elif key == "image_tokens":
+            kw["image_tokens"] = None if x is None else _expect_uint(x, expected="u32")
+        elif key == "audio_tokens":
+            kw["audio_tokens"] = None if x is None else _expect_uint(x, expected="u32")
+        elif key == "video_tokens":
+            kw["video_tokens"] = None if x is None else _expect_uint(x, expected="u32")
     return GenerateMetaInfo(**kw)
 
 
@@ -1809,6 +1818,12 @@ def encode_GenerateMetaInfo(v: GenerateMetaInfo) -> Dict[str, Any]:
         d["weight_version"] = v.weight_version
     if v.weight_versions:
         d["weight_versions"] = [encode_WeightVersionSpan(e) for e in v.weight_versions]
+    if v.image_tokens is not None:
+        d["image_tokens"] = v.image_tokens
+    if v.audio_tokens is not None:
+        d["audio_tokens"] = v.audio_tokens
+    if v.video_tokens is not None:
+        d["video_tokens"] = v.video_tokens
     return d
 
 

python/sglang/srt/managers/scheduler_components/output_streamer.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from dataclasses import dataclass, field
 from typing import (
     TYPE_CHECKING,
@@ -38,6 +39,7 @@ from sglang.srt.speculative.spec_info import SpeculativeAlgorithm
 from sglang.srt.utils.weight_versions import compute_weight_version_spans
 
 if TYPE_CHECKING:
+    from sglang.srt.observability.req_time_stats import SchedulerReqTimeStats
     from sglang.srt.rust_server.server import RustServer
 
 
@@ -47,6 +49,20 @@ logger = logging.getLogger(__name__)
 DEFAULT_FORCE_STREAM_INTERVAL = envs.SGLANG_FORCE_STREAM_INTERVAL.get()
 
 
+def _scheduler_decode_latency(time_stats: SchedulerReqTimeStats) -> Optional[float]:
+    """A finished non-streaming request's decode latency as Python's
+    TokenizerManager measures it: from the scheduler's prefill-finished stamp
+    (`_get_scheduler_first_token_time`, which has the stamp only when metrics
+    are on) to now, its terminal output (`get_decode_latency`). None where it
+    falls back to its own first-output time, which the Rust frontend
+    measures the same way."""
+    stamp = time_stats.prefill_finished_time
+    now = time.perf_counter()
+    if time_stats.enable_metrics and 0.0 < stamp <= now:
+        return now - stamp
+    return None
+
+
 @dataclass(kw_only=True, slots=True)
 class SchedulerOutputStreamer:
     has_additional_customized_info: ClassVar[bool] = False
@@ -214,7 +230,7 @@ class SchedulerOutputStreamer:
         )
         if payload is not None:
             if self.rust_server is not None:
-                self.rust_server.push_generation(payload)
+                self.rust_server.push_generation(payload, acc.decode_latencies)
             else:
                 self.send_to_detokenizer.send_output(payload)
 
@@ -390,6 +406,9 @@ class _GenerationStreamAccumulator:
     # output tokens itself and never consumes the scheduler's incremental-detok
     # offsets (decode_ids / read_offset), so that per-step bookkeeping is skipped.
     rust_server_mode: bool = False
+    # Rust server mode: by row, the decode latency of a finished non-streaming
+    # request, whose first token the Rust frontend does not see.
+    decode_latencies: dict = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if self.return_hidden_states:
@@ -527,6 +546,10 @@ class _GenerationStreamAccumulator:
                     num_output_tokens=len(output_ids_),
                 )
             )
+            if self.rust_server_mode and not req.stream:
+                latency = _scheduler_decode_latency(req.time_stats)
+                if latency is not None:
+                    self.decode_latencies[len(self.rids) - 1] = latency
         else:
             self.weight_versions.append(None)
 

python/sglang/srt/managers/tokenizer_manager.py

@@ -137,6 +137,7 @@ from sglang.srt.observability.metrics_collector import (
     STAT_LOGGER_ROLE_TOKENIZER,
     TokenizerMetricsCollector,
     resolve_collector_class,
+    tokenizer_metric_labels,
 )
 from sglang.srt.observability.req_time_stats import (
     APIServerReqTimeStats,
@@ -834,23 +835,7 @@ class TokenizerManager(TokenizerControlMixin, TokenizerManagerScoreMixin):
     def init_metric_collector_watchdog(self):
         # Metrics
         if self.enable_metrics:
-            engine_type = DisaggregationMode.to_engine_type(
-                get_disagg().disaggregation_mode
-            )
-
-            labels = {
-                "model_name": get_serving().served_model_name,
-                "engine_type": engine_type,
-            }
-            if self.enable_priority_scheduling:
-                labels["priority"] = ""
-            if get_observability().tokenizer_metrics_allowed_custom_labels:
-                for (
-                    label
-                ) in get_observability().tokenizer_metrics_allowed_custom_labels:
-                    labels[label] = ""
-            if get_observability().extra_metric_labels:
-                labels.update(get_observability().extra_metric_labels)
+            labels = tokenizer_metric_labels()
             tokenizer_collector_cls = resolve_collector_class(
                 STAT_LOGGER_ROLE_TOKENIZER,
                 TokenizerMetricsCollector,

python/sglang/srt/observability/metrics_collector.py

@@ -1775,6 +1775,24 @@ class SchedulerMetricsCollector(_StatLoggerDIMixin):
         )
 
 
+def tokenizer_metric_labels() -> Dict[str, str]:
+    """The labels on every ``TokenizerMetricsCollector`` metric. The priority
+    and custom labels start empty; each request fills them in."""
+    labels = {
+        "model_name": get_serving().served_model_name,
+        "engine_type": DisaggregationMode.to_engine_type(
+            get_disagg().disaggregation_mode
+        ),
+    }
+    if get_schedule().enable_priority_scheduling:
+        labels["priority"] = ""
+    for label in get_observability().tokenizer_metrics_allowed_custom_labels or ():
+        labels[label] = ""
+    if get_observability().extra_metric_labels:
+        labels.update(get_observability().extra_metric_labels)
+    return labels
+
+
 class TokenizerMetricsCollector(_StatLoggerDIMixin):
     def __init__(
         self,

python/sglang/srt/rust_server/config.py

@@ -17,6 +17,8 @@ from sglang.srt.entrypoints.v1_loads import (
 from sglang.srt.environ import envs
 from sglang.srt.managers.load_snapshot import ShmLoadSnapshotWriter
 from sglang.srt.managers.utils import compute_num_reserved_tokens
+from sglang.srt.observability.metrics_collector import tokenizer_metric_labels
+from sglang.srt.observability.utils import generate_buckets
 from sglang.srt.runtime_context import (
     get_device,
     get_disagg,
@@ -196,13 +198,34 @@ def _build_server_args(
         load_snapshot_dp_rank=shm_writer.dp_rank if shm_writer else 0,
         accelerator=_accelerator_name(),
         num_accelerators=_num_accelerators_per_dp_rank(),
+        metrics=(
+            _build_metrics_config(ext) if get_observability().enable_metrics else None
+        ),
+    )
+
+
+def _build_metrics_config(ext: ModuleType):
+    """What the Rust server needs to record ``TokenizerMetricsCollector``'s
+    families as Python does, and where every process's samples live."""
+    obs = get_observability()
+
+    def token_buckets(rule: Optional[List[str]]) -> Optional[List[float]]:
+        # The default rule keeps the Rust side's copy of Python's defaults.
+        if not rule or rule[0] == "default":
+            return None
+        return generate_buckets(rule, [])
+
+    return ext.MetricsConfig(
         # Every process writes its prometheus_client samples under this
         # directory (set before the schedulers start); `/metrics` merges them.
-        metrics=(
-            ext.MetricsConfig(multiproc_dir=os.environ["PROMETHEUS_MULTIPROC_DIR"])
-            if get_observability().enable_metrics
-            else None
-        ),
+        multiproc_dir=os.environ["PROMETHEUS_MULTIPROC_DIR"],
+        labels=tokenizer_metric_labels(),
+        priority_label=get_schedule().enable_priority_scheduling,
+        bucket_time_to_first_token=obs.bucket_time_to_first_token,
+        bucket_inter_token_latency=obs.bucket_inter_token_latency,
+        bucket_e2e_request_latency=obs.bucket_e2e_request_latency,
+        bucket_prompt_tokens=token_buckets(obs.prompt_tokens_buckets),
+        bucket_generation_tokens=token_buckets(obs.generation_tokens_buckets),
     )
 
 

python/sglang/srt/rust_server/server.py

@@ -51,6 +51,11 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# The extras families' header columns (`BatchHeader.out_lp_lens` through
+# `hidden_poslens`). A frame without extras that sends a later column pads
+# them empty; `push_generation` asserts the count whenever it sends extras.
+_EXTRAS_HEADER_COLS = 12
+
 
 class RustServer:
     """Owns the embedded multi-threaded Rust server (``sglang_server.Server``).
@@ -89,6 +94,12 @@ class RustServer:
         clamps to what the context window leaves after the prompt."""
         return SamplingParams().max_new_tokens
 
+    @classmethod
+    def _server_kwargs(cls, scheduler: Scheduler) -> Dict[str, Any]:
+        """Keyword arguments a model package's own ``Server`` class takes
+        beyond the shared ones (see ``_load_extension``)."""
+        return {}
+
     @classmethod
     def _load_extension(cls) -> ModuleType:
         from sglang.srt.rust_extensions import load_rust_extension
@@ -197,6 +208,7 @@ class RustServer:
             # None runs unpinned; otherwise the list carries the pinning decision.
             cores=server_cores,
             port_offset=local_dp_rank,
+            **cls._server_kwargs(scheduler),
         )
         instance = cls(server, http_port=listen_port)
 
@@ -389,21 +401,33 @@ class RustServer:
         Python's ``TokenizerManager._handle_abort_req`` builds from the same
         ``AbortReq``: a one-request batch carrying no new tokens and the abort
         finish reason. The counts are the request's own, so the frame does not
-        reset the ones a stream already reported."""
+        reset the ones a stream already reported. The frame is marked as ended
+        outside the stream: like ``_handle_abort_req``, the frontend records no
+        request metrics for it."""
         finish_reason = abort.finished_reason or {
             "type": "abort",
             "message": abort.abort_message or "Abort in waiting queue",
         }
-        header = msgspec.msgpack.encode(
-            [
-                [abort.rid],
-                [finish_reason],
-                [len(req.origin_input_ids)],
-                [0],
-                [req.cached_tokens],
-                [req.reasoning_tokens],
-            ]
-        )
+        header_cols = [
+            [abort.rid],
+            [finish_reason],
+            [len(req.origin_input_ids)],
+            [0],
+            [req.cached_tokens],
+            [req.reasoning_tokens],
+        ]
+        header_cols += [[]] * _EXTRAS_HEADER_COLS
+        header_cols += [
+            [abort.weight_versions] if abort.weight_versions is not None else [],
+            [],  # image_tokens
+            [],  # audio_tokens
+            [],  # video_tokens
+            [],  # decode_latencies
+            # Ended outside the output stream: `_handle_abort_req` records no
+            # request metrics for it.
+            [True],
+        ]
+        header = msgspec.msgpack.encode(header_cols)
         if not self.server.push_decode_result_batch(header, [b""]):
             logger.warning(
                 "Rust egress closed; dropped abort of %s during shutdown", abort.rid
@@ -413,7 +437,9 @@ class RustServer:
         """Acknowledge the health probes skipped while the scheduler was busy."""
         self.server.push_health_ack()
 
-    def push_generation(self, payload: BatchTokenIDOutput) -> None:
+    def push_generation(
+        self, payload: BatchTokenIDOutput, decode_latencies: Dict[int, float]
+    ) -> None:
         """Egress redirect for generation output (replaces the zmq detokenizer).
 
         Push the WHOLE batch into the Rust egress ring as one frame (-> detokenizer
@@ -425,7 +451,10 @@ class RustServer:
             cached_tokens, reasoning_tokens``) plus
             the shape metadata for the optional families (``*_lens`` element counts
             for the flat logprob columns, ``*_reqlens``/``*_poslens`` for the ragged
-            and hidden ones).
+            and hidden ones), then the finished requests' weight-version
+            spans, a multimodal model's media token counts, and
+            ``decode_latencies`` (by row: a finished non-streaming request's
+            decode latency in seconds).
           - ``data``: the raw little-endian numeric buffer: the token ids
             first, as the 8-byte ``array("q")`` bytes the scheduler already
             holds, then every logprob / hidden column as 4-byte elements
@@ -563,9 +592,39 @@ class RustServer:
                 for i in range(batch_size):
                     accept(i)
 
+            extras_start = len(header_cols)
             for extra in extras:
                 header_cols += extra.header_cols()
                 data_cols += extra.data_cols()
+            # The padding below must line up with these columns.
+            assert len(header_cols) - extras_start == _EXTRAS_HEADER_COLS, (
+                "_EXTRAS_HEADER_COLS no longer matches the extras families"
+            )
+
+        # After the extras: the finished requests' weight-version spans (None
+        # for the others), a multimodal model's media token counts and the
+        # finished requests' decode latencies, each empty when nothing in the
+        # batch carries it. Only the last is built here, for finished rows; the
+        # header stops at its last non-empty column.
+        latencies = []
+        if decode_latencies:
+            latencies = [None] * len(rids)
+            for row, latency in decode_latencies.items():
+                latencies[row] = latency
+        media = self._multimodal_enabled
+        trailing = [
+            payload.weight_versions or [],
+            (payload.image_tokens or []) if media else [],
+            (payload.audio_tokens or []) if media else [],
+            (payload.video_tokens or []) if media else [],
+            latencies,
+        ]
+        while trailing and not trailing[-1]:
+            trailing.pop()
+        if trailing:
+            if not has_extra:
+                header_cols += [[]] * _EXTRAS_HEADER_COLS
+            header_cols += trailing
 
         header = msgspec.msgpack.encode(header_cols)
         # Pass the raw column list; the Rust side concatenates it into the frame

rust/sglang-api-types/src/generated/sglang.api.v1.rs

@@ -590,6 +590,13 @@ pub struct GenerateMetaInfo {
     /// Omitted until terminal output; reported spans are always nonempty.
     #[prost(message, repeated, tag = "17")]
     pub weight_versions: ::prost::alloc::vec::Vec<WeightVersionSpan>,
+    /// Media prompt tokens per modality, on every frame; omitted when zero.
+    #[prost(uint32, optional, tag = "18")]
+    pub image_tokens: ::core::option::Option<u32>,
+    #[prost(uint32, optional, tag = "19")]
+    pub audio_tokens: ::core::option::Option<u32>,
+    #[prost(uint32, optional, tag = "20")]
+    pub video_tokens: ::core::option::Option<u32>,
 }
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct WeightVersionSpan {

rust/sglang-api-types/src/generated/sglang.api.v1.serde.rs

@@ -2618,6 +2618,15 @@ impl ::serde::Serialize for GenerateMetaInfo {
                 &self.weight_versions,
             )?;
         }
+        if let Some(v) = &self.image_tokens {
+            ::serde::ser::SerializeMap::serialize_entry(&mut map, "image_tokens", v)?;
+        }
+        if let Some(v) = &self.audio_tokens {
+            ::serde::ser::SerializeMap::serialize_entry(&mut map, "audio_tokens", v)?;
+        }
+        if let Some(v) = &self.video_tokens {
+            ::serde::ser::SerializeMap::serialize_entry(&mut map, "video_tokens", v)?;
+        }
         ::serde::ser::SerializeMap::end(map)
     }
 }
@@ -2651,6 +2660,9 @@ impl<'de> ::serde::Deserialize<'de> for GenerateMetaInfo {
                 let mut f_output_token_logprobs_length = None;
                 let mut f_weight_version = None;
                 let mut f_weight_versions = None;
+                let mut f_image_tokens = None;
+                let mut f_audio_tokens = None;
+                let mut f_video_tokens = None;
                 while let Some(key) = map.next_key::<::std::borrow::Cow<'_, str>>()? {
                     match key.as_ref() {
                         "id" => {
@@ -2794,6 +2806,24 @@ impl<'de> ::serde::Deserialize<'de> for GenerateMetaInfo {
                                     .unwrap_or_default(),
                             );
                         }
+                        "image_tokens" => {
+                            if f_image_tokens.is_some() {
+                                return Err(::serde::de::Error::duplicate_field("image_tokens"));
+                            }
+                            f_image_tokens = Some(map.next_value::<Option<u32>>()?);
+                        }
+                        "audio_tokens" => {
+                            if f_audio_tokens.is_some() {
+                                return Err(::serde::de::Error::duplicate_field("audio_tokens"));
+                            }
+                            f_audio_tokens = Some(map.next_value::<Option<u32>>()?);
+                        }
+                        "video_tokens" => {
+                            if f_video_tokens.is_some() {
+                                return Err(::serde::de::Error::duplicate_field("video_tokens"));
+                            }
+                            f_video_tokens = Some(map.next_value::<Option<u32>>()?);
+                        }
                         _ => {
                             map.next_value::<::serde::de::IgnoredAny>()?;
                         }
@@ -2817,6 +2847,9 @@ impl<'de> ::serde::Deserialize<'de> for GenerateMetaInfo {
                     output_token_logprobs_length: f_output_token_logprobs_length.flatten(),
                     weight_version: f_weight_version.flatten(),
                     weight_versions: f_weight_versions.unwrap_or_default(),
+                    image_tokens: f_image_tokens.flatten(),
+                    audio_tokens: f_audio_tokens.flatten(),
+                    video_tokens: f_video_tokens.flatten(),
                 })
             }
         }

rust/sglang-api-types/testdata/json_parity.json

@@ -238,7 +238,7 @@
     },
     {
       "type": "GenerateResponse",
-      "name": "terminal batch frame with every logprob family, null rows, hidden states, e2e_latency, cache and reasoning counts, index",
+      "name": "terminal batch frame with every logprob family, null rows, hidden states, e2e_latency, cache and reasoning counts, weight versions, media counts, index",
       "input": {
         "text": "ok",
         "meta_info": {
@@ -305,14 +305,28 @@
           ],
           "e2e_latency": 0.01,
           "cached_tokens": 4,
-          "reasoning_tokens": 2
+          "reasoning_tokens": 2,
+          "weight_version": "new",
+          "weight_versions": [
+            {
+              "version": "old",
+              "start": 0,
+              "end": 1
+            },
+            {
+              "version": "new",
+              "start": 1,
+              "end": 1
+            }
+          ],
+          "image_tokens": 8
         },
         "output_ids": [
           5
         ],
         "index": 0
       },
-      "canonical": "{\"text\":\"ok\",\"meta_info\":{\"id\":\"r\",\"prompt_tokens\":1,\"completion_tokens\":1,\"finish_reason\":{\"type\":\"stop\",\"matched\":[1,2]},\"output_token_logprobs\":[[-0.5,5,\"x\"],[null,6,null]],\"input_token_logprobs\":[],\"output_top_logprobs\":[null,[[-0.25,6,null]]],\"input_top_logprobs\":[[[-0.125,1,\"<s>\"]]],\"output_token_ids_logprobs\":[[[-0.5,9,null]]],\"input_token_ids_logprobs\":[null],\"hidden_states\":[[0.5,0.25],[]],\"e2e_latency\":0.01,\"cached_tokens\":4,\"reasoning_tokens\":2},\"output_ids\":[5],\"index\":0}"
+      "canonical": "{\"text\":\"ok\",\"meta_info\":{\"id\":\"r\",\"prompt_tokens\":1,\"completion_tokens\":1,\"finish_reason\":{\"type\":\"stop\",\"matched\":[1,2]},\"output_token_logprobs\":[[-0.5,5,\"x\"],[null,6,null]],\"input_token_logprobs\":[],\"output_top_logprobs\":[null,[[-0.25,6,null]]],\"input_top_logprobs\":[[[-0.125,1,\"<s>\"]]],\"output_token_ids_logprobs\":[[[-0.5,9,null]]],\"input_token_ids_logprobs\":[null],\"hidden_states\":[[0.5,0.25],[]],\"e2e_latency\":0.01,\"cached_tokens\":4,\"reasoning_tokens\":2,\"weight_version\":\"new\",\"weight_versions\":[{\"version\":\"old\",\"start\":0,\"end\":1},{\"version\":\"new\",\"start\":1,\"end\":1}],\"image_tokens\":8},\"output_ids\":[5],\"index\":0}"
     },
     {
       "type": "GenerateResponse",

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

@@ -12,9 +12,13 @@ mod error;
 mod frame;
 mod handle;
 mod hicache;
+mod metrics;
 mod prefetch;
 
 pub(crate) use error::{CoreError, CoreErrorKind};
+#[cfg(test)]
+pub(crate) use frame::frame_metadata_case;
 pub(crate) use frame::{CoreEvent, CoreOutput, HealthStatus};
 pub(crate) use handle::{CoreCall, CoreConfig, CoreHandle, CoreMetadata, recv_indexed};
 pub(crate) use hicache::HiCacheManager;
+pub(crate) use metrics::{FLUSH_INTERVAL, FrontendMetrics};

rust/sglang-server/src/api_server/core/frame.rs

@@ -5,11 +5,13 @@
 //! as protobuf, so a column added here reaches both the same way. No Axum,
 //! SSE, Tonic, or scheduler-channel details.
 
+use std::sync::Arc;
+
 use sglang_api_types::api::v1 as api;
 
 use super::CoreError;
 use crate::message::finish_reason::{FinishKind, FinishReason, Matched};
-use crate::message::response::{ChunkEvent, ChunkExtras};
+use crate::message::response::{ChunkEvent, ChunkExtras, WeightVersionSpan};
 use crate::message::types::TokenIds;
 
 /// Canonical transport-neutral output payload for generation events.
@@ -26,6 +28,9 @@ pub(crate) struct CoreOutput {
     pub(crate) prompt_tokens: u32,
     pub(crate) cached_tokens: u32,
     pub(crate) reasoning_tokens: u64,
+    /// The launch weight version, which frames report until a terminal one's
+    /// spans name the last; stamped by the core call.
+    pub(crate) weight_version: Option<Arc<str>>,
     pub(crate) text: String,
     pub(crate) completion_tokens: u64,
     pub(crate) stop_token_trimmed: bool,
@@ -52,6 +57,7 @@ impl From<ChunkEvent> for CoreOutput {
             prompt_tokens,
             cached_tokens,
             reasoning_tokens,
+            weight_version: None,
             text,
             completion_tokens,
             stop_token_trimmed,
@@ -87,6 +93,9 @@ impl CoreOutput {
             self.finish_reason = delta.finish_reason.clone();
             self.stop_token_trimmed = delta.stop_token_trimmed;
         }
+        if delta.weight_version.is_some() {
+            self.weight_version = delta.weight_version.clone();
+        }
 
         let Some(delta_extras) = delta.extras.as_deref() else {
             return;
@@ -153,6 +162,20 @@ impl CoreOutput {
             extras.hidden_val = delta_extras.hidden_val.clone();
             extras.hidden_lens = delta_extras.hidden_lens.clone();
         }
+        if !delta_extras.weight_versions.is_empty() {
+            extras.weight_versions = delta_extras.weight_versions.clone();
+        }
+        // A request's media counts never change; a frame without them (a
+        // scheduler abort) keeps the ones the stream reported.
+        for (count, delta_count) in [
+            (&mut extras.image_tokens, delta_extras.image_tokens),
+            (&mut extras.audio_tokens, delta_extras.audio_tokens),
+            (&mut extras.video_tokens, delta_extras.video_tokens),
+        ] {
+            if delta_count != 0 {
+                *count = delta_count;
+            }
+        }
     }
 }
 
@@ -185,6 +208,16 @@ impl CoreEvent {
 }
 
 impl CoreOutput {
+    /// The scheduler's decode latency for a finished non-streaming request,
+    /// whose first token this frontend does not see. It rides only that
+    /// request's terminal delta and is read off it there (for a model package's
+    /// [`GenerateHook`](crate::GenerateHook)), so [`append_delta`](Self::append_delta)
+    /// does not fold it into the cumulative output.
+    pub(crate) fn scheduler_decode_latency(&self) -> Option<std::time::Duration> {
+        let seconds = self.extras.as_deref()?.decode_latency?;
+        std::time::Duration::try_from_secs_f64(seconds).ok()
+    }
+
     /// One `/generate` frame. `index` is the batch position (batch form only);
     /// `e2e_latency` rides the terminal frame only.
     ///
@@ -220,12 +253,23 @@ impl CoreOutput {
             e2e_latency,
             cached_tokens: self.cached_tokens,
             reasoning_tokens: self.reasoning_tokens,
+            weight_version: self.weight_version.as_deref().map(str::to_owned),
             ..Default::default()
         };
         let Some(extras) = self.extras.as_deref() else {
             return meta;
         };
 
+        meta.weight_versions =
+            visible_weight_versions(&extras.weight_versions, self.completion_tokens);
+        if let Some(span) = meta.weight_versions.last() {
+            meta.weight_version = Some(span.version.clone());
+        }
+        let nonzero = |count: u32| (count != 0).then_some(count);
+        meta.image_tokens = nonzero(extras.image_tokens);
+        meta.audio_tokens = nonzero(extras.audio_tokens);
+        meta.video_tokens = nonzero(extras.video_tokens);
+
         // Python always sets input and output token logprobs together, including
         // the empty input list emitted by a disaggregated decode node.
         if !extras.out_lp_val.is_empty() || !extras.in_lp_val.is_empty() {
@@ -279,6 +323,24 @@ impl CoreOutput {
     }
 }
 
+/// Python's `add_weight_versions_to_meta_info`: the spans that start inside
+/// the completion (and the first one, even for an empty completion), each
+/// clipped to it.
+fn visible_weight_versions(
+    spans: &[WeightVersionSpan],
+    completion_tokens: u64,
+) -> Vec<api::WeightVersionSpan> {
+    spans
+        .iter()
+        .filter(|span| span.start == 0 || span.start < completion_tokens)
+        .map(|span| api::WeightVersionSpan {
+            version: span.version.clone(),
+            start: span.start as i64,
+            end: span.end.min(completion_tokens) as i64,
+        })
+        .collect()
+}
+
 /// A decoded-text column becomes the entries' text source only when populated.
 fn opt_texts(texts: &[String]) -> Option<&[String]> {
     (!texts.is_empty()).then_some(texts)
@@ -394,6 +456,29 @@ fn matched_value(matched: &Matched) -> api::Matched {
     api::Matched { value: Some(value) }
 }
 
+/// One case of `testdata/frame_metadata.json` (built by Python's
+/// `build_rust_frame_metadata_fixture`), and the events its frame decodes to.
+#[cfg(test)]
+pub(crate) fn frame_metadata_case(name: &str) -> (serde_json::Value, Vec<ChunkEvent>) {
+    use crate::message::response::{for_each_chunk, frame_decode_batch_cols};
+
+    let fixture: serde_json::Value =
+        serde_json::from_str(include_str!("../../../testdata/frame_metadata.json")).unwrap();
+    let case = fixture[name].clone();
+    let bytes = |hex: &serde_json::Value| -> Vec<u8> {
+        let hex = hex.as_str().unwrap_or_default();
+        (0..hex.len())
+            .step_by(2)
+            .map(|i| u8::from_str_radix(&hex[i..i + 2], 16).unwrap())
+            .collect()
+    };
+    let data = bytes(&case["data"]);
+    let framed = frame_decode_batch_cols(&bytes(&case["header"]), &[data.as_slice()]);
+    let mut events = Vec::new();
+    assert!(for_each_chunk(&framed[1..], |event| events.push(event)).ok);
+    (case, events)
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -461,6 +546,79 @@ mod tests {
         assert_eq!(opt_texts(&texts), Some(texts.as_slice()));
     }
 
+    /// The weight-version and media-count columns, as Python's
+    /// `push_generation` and `push_abort` send them, render the `meta_info`
+    /// keys Python's TokenizerManager derives from the same values
+    /// (`testdata/frame_metadata.json`, built by
+    /// `build_rust_frame_metadata_fixture`): spans on a terminal frame, clipped
+    /// to the completion the client received, the launch version before one,
+    /// and only nonzero media counts, also behind a frame's logprob columns;
+    /// the scheduler's decode latency reaches its row, and only the abort is
+    /// marked as ended outside the output stream.
+    #[test]
+    fn frame_metadata_matches_python() {
+        let keys = |output: &CoreOutput| {
+            let meta = json(&output.meta_info("rid", None));
+            let mut keys = serde_json::Map::new();
+            for key in [
+                "weight_version",
+                "weight_versions",
+                "image_tokens",
+                "audio_tokens",
+                "video_tokens",
+            ] {
+                if let Some(value) = meta.get(key) {
+                    keys.insert(key.into(), value.clone());
+                }
+            }
+            serde_json::Value::Object(keys)
+        };
+        let launch = Some(Arc::<str>::from("launch"));
+
+        for name in ["batch", "logprob_batch"] {
+            let (batch, events) = super::frame_metadata_case(name);
+            assert_eq!(events.len(), 2);
+            for event in events {
+                let rid = event.rid.as_str().to_owned();
+                assert!(!event.extras.as_deref().is_some_and(|e| e.abort_output));
+                let completion_tokens = event.token_ids.len() as u64;
+                let mut output = CoreOutput::from(event);
+                output.weight_version = launch.clone();
+                output.completion_tokens = completion_tokens;
+                assert_eq!(keys(&output), batch["meta_info"][&rid], "{name} {rid}");
+                assert_eq!(
+                    output.scheduler_decode_latency(),
+                    batch["decode_latencies"][&rid]
+                        .as_f64()
+                        .map(std::time::Duration::from_secs_f64),
+                    "{name} {rid}"
+                );
+                if rid == "logprobs" {
+                    let meta = json(&output.meta_info(&rid, None));
+                    assert_eq!(
+                        meta["output_token_logprobs"],
+                        serde_json::json!([[-0.5, 7, null], [-0.25, 8, null]])
+                    );
+                }
+            }
+        }
+
+        // A scheduler abort folds into what the stream already sent, and its
+        // spans are clipped to those tokens.
+        let (abort, mut events) = super::frame_metadata_case("abort");
+        let mut aborted = CoreOutput {
+            weight_version: launch.clone(),
+            completion_tokens: abort["completion_tokens"].as_u64().unwrap(),
+            ..Default::default()
+        };
+        let event = events.pop().unwrap();
+        assert!(event.extras.as_deref().is_some_and(|e| e.abort_output));
+        let mut delta = CoreOutput::from(event);
+        delta.weight_version = launch;
+        aborted.append_delta(&delta);
+        assert_eq!(keys(&aborted), abort["meta_info"]);
+    }
+
     /// A trimmed stop token stays in the accumulated `token_ids` but out of the
     /// parser view, which must line up with the accumulated text. A stop token
     /// kept in the text (`no_stop_trim`) stays in both.

rust/sglang-server/src/api_server/core/handle.rs

@@ -8,7 +8,7 @@ use std::sync::{
     Arc, Mutex, OnceLock,
     atomic::{AtomicBool, Ordering},
 };
-use std::time::Duration;
+use std::time::{Duration, Instant};
 
 use sglang_api_types::api::v1 as api;
 use tokio::sync::{mpsc, oneshot};
@@ -28,6 +28,7 @@ use crate::tokenizer_manager::wiring::{AbortSource, RequestAdmission, TmEvent};
 use crate::utils::error::Error;
 use crate::utils::fsm::RequestState;
 
+use super::metrics::{FrontendMetrics, RequestMetrics};
 use super::{CoreError, CoreEvent, HealthStatus, prefetch};
 use crate::message::info::{InternalState, ModelInfo, ServerInfo};
 
@@ -42,6 +43,8 @@ pub(crate) struct CoreConfig {
     pub(crate) is_disaggregation: bool,
     pub(crate) mm_limits: Vec<(String, usize)>,
     pub(crate) metadata: CoreMetadata,
+    /// Request metrics, when `--enable-metrics` is on.
+    pub(crate) metrics: Option<Arc<FrontendMetrics>>,
 }
 
 /// Immutable, transport-independent metadata retained by the core.
@@ -53,7 +56,7 @@ pub(crate) struct CoreMetadata {
     served_model_name: String,
     tokenizer_path: String,
     preferred_sampling_params: Option<crate::message::config::PreferredSamplingParams>,
-    weight_version: Option<String>,
+    weight_version: Option<Arc<str>>,
     load_format: Option<String>,
     reasoning_parser: Option<String>,
     tool_call_parser: Option<String>,
@@ -75,7 +78,7 @@ impl From<&ServerArgs> for CoreMetadata {
             served_model_name: args.served_model_name.clone(),
             tokenizer_path: args.tokenizer_path.clone(),
             preferred_sampling_params: args.preferred_sampling_params.clone(),
-            weight_version: args.weight_version.clone(),
+            weight_version: args.weight_version.as_deref().map(Arc::from),
             load_format: args.load_format.clone(),
             reasoning_parser: args.reasoning_parser.clone(),
             tool_call_parser: args.tool_call_parser.clone(),
@@ -114,6 +117,7 @@ struct CoreInner {
     mm_processor: OnceLock<Arc<dyn MmProcessor>>,
     metadata: CoreMetadata,
     generations: Generations,
+    metrics: Option<Arc<FrontendMetrics>>,
 }
 
 impl CoreHandle {
@@ -134,6 +138,7 @@ impl CoreHandle {
                 mm_processor: OnceLock::new(),
                 metadata: config.metadata,
                 generations: Generations::default(),
+                metrics: config.metrics,
             }),
         }
     }
@@ -158,10 +163,16 @@ impl CoreHandle {
         self.inner.response_activity.load(Ordering::Relaxed)
     }
 
+    /// The request metrics, when enabled.
+    pub(crate) fn metrics(&self) -> Option<&Arc<FrontendMetrics>> {
+        self.inner.metrics.as_ref()
+    }
+
     pub(crate) async fn generate(
         &self,
         mut request: GenerateRequest,
     ) -> Result<CoreCall, CoreError> {
+        let created_at = Instant::now();
         let mut generation = self
             .inner
             .generations
@@ -171,19 +182,30 @@ impl CoreHandle {
             _ = generation.cancelled() => return Err(CoreError::Cancelled("request aborted".into())),
             result = self.prepare_generations(std::slice::from_mut(&mut request)) => result?,
         }
+        let metrics = self.request_metrics(&request, created_at);
         self.submit_generation(request, generation)
             .await
-            .map(CoreCall::new)
+            .map(|runtime| self.generation_call(runtime, metrics))
     }
 
     /// Submit core-owned work without exposing it through the public Abort
     /// registry. Drop still owns cancellation through `RuntimeCall`.
-    async fn generate_internal(&self, mut request: GenerateRequest) -> Result<CoreCall, CoreError> {
+    /// `log_metrics` is Python's flag of the same name: the health probe's
+    /// request is not counted.
+    async fn generate_internal(
+        &self,
+        mut request: GenerateRequest,
+        log_metrics: bool,
+    ) -> Result<CoreCall, CoreError> {
+        let created_at = Instant::now();
         self.prepare_generations(std::slice::from_mut(&mut request))
             .await?;
+        let metrics = log_metrics
+            .then(|| self.request_metrics(&request, created_at))
+            .flatten();
         self.submit(RequestKind::Generate(Box::new(request)))
             .await
-            .map(CoreCall::new)
+            .map(|runtime| self.generation_call(runtime, metrics))
     }
 
     /// Prepare the group before submitting any surviving request. Explicit
@@ -194,6 +216,7 @@ impl CoreHandle {
         &self,
         mut requests: Vec<GenerateRequest>,
     ) -> Result<Vec<CoreCall>, CoreError> {
+        let created_at = Instant::now();
         if requests.is_empty() {
             return Ok(Vec::new());
         }
@@ -222,8 +245,9 @@ impl CoreHandle {
         let mut calls = Vec::with_capacity(requests.len());
         for (request, generation) in requests.into_iter().zip(generations) {
             let rid = request.rid.clone();
+            let metrics = self.request_metrics(&request, created_at);
             let call = match self.submit_generation(request, generation).await {
-                Ok(call) => CoreCall::new(call),
+                Ok(call) => self.generation_call(call, metrics),
                 // Cancellation may happen during preparation or while waiting
                 // for intake. Keep this slot without rolling back siblings.
                 Err(CoreError::Cancelled(_)) => {
@@ -252,6 +276,19 @@ impl CoreHandle {
             .await
     }
 
+    fn generation_call(&self, runtime: RuntimeCall, metrics: Option<RequestMetrics>) -> CoreCall {
+        CoreCall::new(runtime, self.inner.metadata.weight_version.clone(), metrics)
+    }
+
+    fn request_metrics(
+        &self,
+        request: &GenerateRequest,
+        created_at: Instant,
+    ) -> Option<RequestMetrics> {
+        let metrics = self.inner.metrics.as_ref()?;
+        Some(metrics.start_request(request, created_at))
+    }
+
     async fn prepare_generations(&self, requests: &mut [GenerateRequest]) -> Result<(), CoreError> {
         prefetch::prefetch_all(
             requests,
@@ -274,7 +311,7 @@ impl CoreHandle {
             tokenizer_path: metadata.tokenizer_path.clone(),
             is_generation: true,
             preferred_sampling_params: metadata.preferred_sampling_params.clone(),
-            weight_version: metadata.weight_version.clone(),
+            weight_version: metadata.weight_version.as_deref().map(str::to_owned),
             load_format: metadata.load_format.clone(),
             reasoning_parser: metadata.reasoning_parser.clone(),
             tool_call_parser: metadata.tool_call_parser.clone(),
@@ -468,7 +505,7 @@ impl CoreHandle {
         // The probe is intentionally not drained. Its call stays alive while
         // the heartbeat is observed, then aborts on drop to clean up a probe
         // that a busy scheduler skipped without producing a terminal frame.
-        let _probe = self.generate_internal(probe).await?;
+        let _probe = self.generate_internal(probe, false).await?;
 
         let deadline = tokio::time::Instant::now() + timeout;
         loop {
@@ -491,23 +528,26 @@ impl CoreHandle {
             (Some("The capital city of France is".to_string()), None)
         };
         let mut call = self
-            .generate_internal(GenerateRequest {
-                rid: Rid::new(),
-                text,
-                input_ids,
-                sampling_params: SamplingParams {
-                    max_new_tokens: Some(8),
-                    temperature: 0.0,
+            .generate_internal(
+                GenerateRequest {
+                    rid: Rid::new(),
+                    text,
+                    input_ids,
+                    sampling_params: SamplingParams {
+                        max_new_tokens: Some(8),
+                        temperature: 0.0,
+                        ..Default::default()
+                    },
+                    stream: false,
+                    bootstrap_host: self
+                        .inner
+                        .is_disaggregation
+                        .then(|| FAKE_BOOTSTRAP_HOST.into()),
+                    bootstrap_room: self.inner.is_disaggregation.then_some(0),
                     ..Default::default()
                 },
-                stream: false,
-                bootstrap_host: self
-                    .inner
-                    .is_disaggregation
-                    .then(|| FAKE_BOOTSTRAP_HOST.into()),
-                bootstrap_room: self.inner.is_disaggregation.then_some(0),
-                ..Default::default()
-            })
+                true,
+            )
             .await?;
         loop {
             match call.recv().await {
@@ -600,13 +640,30 @@ fn translate_runtime_error(error: Error) -> CoreError {
 pub(crate) struct CoreCall {
     runtime: RuntimeCall,
     events_finished: bool,
+    /// The launch weight version, stamped on every output (see
+    /// [`CoreOutput::weight_version`](super::CoreOutput)).
+    weight_version: Option<Arc<str>>,
+    /// Observes every runtime item as the call receives it.
+    metrics: Option<RequestMetrics>,
 }
 
 impl CoreCall {
-    fn new(runtime: RuntimeCall) -> Self {
+    fn new(
+        runtime: RuntimeCall,
+        weight_version: Option<Arc<str>>,
+        metrics: Option<RequestMetrics>,
+    ) -> Self {
         Self {
             runtime,
             events_finished: false,
+            weight_version,
+            metrics,
+        }
+    }
+
+    fn observe(&mut self, item: Option<&ResponseItem>) {
+        if let Some(metrics) = &mut self.metrics {
+            metrics.observe(item);
         }
     }
 
@@ -616,15 +673,19 @@ impl CoreCall {
         response_tx
             .try_send(ResponseItem::Error(Error::Disconnected))
             .expect("new response channel has room for its terminal error");
-        Self::new(RuntimeCall {
-            rid,
-            response_rx,
-            abort_tx,
-            admission: RequestAdmission::pending(),
-            expected_response: ExpectedResponse::Generate,
-            in_flight: false,
-            generation: None,
-        })
+        Self::new(
+            RuntimeCall {
+                rid,
+                response_rx,
+                abort_tx,
+                admission: RequestAdmission::pending(),
+                expected_response: ExpectedResponse::Generate,
+                in_flight: false,
+                generation: None,
+            },
+            None,
+            None,
+        )
     }
 
     /// ID that adapters should return to their client.
@@ -640,8 +701,10 @@ impl CoreCall {
             return None;
         }
 
-        let event = match self.runtime.recv().await {
-            Some(item) => generation_event(item),
+        let item = self.runtime.recv().await;
+        self.observe(item.as_ref());
+        let event = match item {
+            Some(item) => self.generation_event(item),
             None => CoreEvent::Failed(CoreError::Internal(
                 "response truncated before completion".into(),
             )),
@@ -655,7 +718,10 @@ impl CoreCall {
     pub(crate) fn drain_ready(&mut self, events: &mut Vec<CoreEvent>) {
         while !self.events_finished {
             let event = match self.runtime.try_recv() {
-                Ok(item) => generation_event(item),
+                Ok(item) => {
+                    self.observe(Some(&item));
+                    self.generation_event(item)
+                }
                 // Preserve every queued event before reporting a premature
                 // close. The next awaited `recv` translates that close into
                 // one terminal failure, maintaining stream order while still
@@ -677,15 +743,35 @@ impl CoreCall {
     ) -> Self {
         let admission = RequestAdmission::pending();
         assert!(admission.try_accept(), "test calls start already admitted");
-        Self::new(RuntimeCall {
-            rid,
-            response_rx,
-            abort_tx,
-            admission,
-            expected_response: ExpectedResponse::Generate,
-            in_flight: true,
-            generation: None,
-        })
+        Self::new(
+            RuntimeCall {
+                rid,
+                response_rx,
+                abort_tx,
+                admission,
+                expected_response: ExpectedResponse::Generate,
+                in_flight: true,
+                generation: None,
+            },
+            None,
+            None,
+        )
+    }
+
+    fn generation_event(&self, item: ResponseItem) -> CoreEvent {
+        let mut event = generation_event(item);
+        if let CoreEvent::Delta(output) | CoreEvent::Finished(output) = &mut event {
+            output.weight_version = self.weight_version.clone();
+        }
+        event
+    }
+}
+
+impl Drop for CoreCall {
+    fn drop(&mut self) {
+        if let Some(metrics) = &self.metrics {
+            metrics.call_dropped();
+        }
     }
 }
 
@@ -1062,6 +1148,7 @@ mod tests {
                 is_disaggregation,
                 mm_limits,
                 metadata: CoreMetadata::default(),
+                metrics: None,
             },
         )
     }
@@ -1094,6 +1181,7 @@ mod tests {
                 startup_ready: false,
                 is_disaggregation: false,
                 mm_limits: Vec::new(),
+                metrics: None,
                 metadata: CoreMetadata {
                     served_model_name: "served-model".into(),
                     max_context_length: 1_u64 << 32,

rust/sglang-server/src/api_server/core/hicache.rs

@@ -162,6 +162,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: CoreMetadata::default(),
+                metrics: None,
             },
         );
         let manager = Arc::new(HiCacheManager::new(core, &ServerArgs::default()));

rust/sglang-server/src/api_server/disaggregation/bootstrap.rs

@@ -479,6 +479,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(server_args),
+            generate_hook: None,
         };
         (
             crate::runtime::start(cfg, None).expect("start runtime"),

rust/sglang-server/src/api_server/grpc/service.rs

@@ -326,6 +326,7 @@ mod tests {
                         },
                         ..Default::default()
                     }),
+                    metrics: None,
                 },
             );
             Self {
@@ -659,6 +660,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: CoreMetadata::default(),
+                metrics: None,
             },
         );
         let service = GrpcService::for_test(core, None, false, Duration::from_secs(1));

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

@@ -1,4 +1,5 @@
 pub(crate) mod app;
+pub(crate) mod generate_hook;
 mod hicache;
 mod info;
 mod loads;

rust/sglang-server/src/api_server/http/app.rs

@@ -7,9 +7,10 @@ use std::time::Duration;
 
 use axum::Router;
 
+use super::generate_hook::GenerateHook;
 use super::{hicache, info, loads, metrics, native_api, openai};
 use crate::HttpLayer;
-use crate::api_server::core::{CoreHandle, HiCacheManager};
+use crate::api_server::core::{CoreHandle, FLUSH_INTERVAL, HiCacheManager};
 use crate::api_server::disaggregation::bootstrap as pd_bootstrap;
 use crate::api_server::log;
 use crate::message::config::ServerArgs;
@@ -28,6 +29,7 @@ pub(super) struct AppState {
     pub(super) server_args: Arc<ServerArgs>,
     pub(super) generation_policy: GenerationPolicy,
     pub(super) chat_formatter: Option<openai::ChatFormatter>,
+    pub(super) generate_hook: Option<Arc<dyn GenerateHook>>,
 }
 
 /// Every listener warms its own rank; health stays 503 until it succeeds,
@@ -64,6 +66,7 @@ pub async fn serve(
     // Whether this listener serves the node's `/metrics`.
     metrics_listener: bool,
     http_layer: Option<HttpLayer>,
+    generate_hook: Option<Arc<dyn GenerateHook>>,
     // The runtime's shutdown signal, shared with every worker stage: it fires
     // (disconnects) when `Runtime::request_shutdown` drops the sender, at
     // which point `serve` stops accepting and its in-flight handlers are
@@ -78,6 +81,7 @@ pub async fn serve(
         server_args: server_args.clone(),
         generation_policy,
         chat_formatter,
+        generate_hook,
     });
     // Cancelled with the api runtime on shutdown.
     tokio::spawn(startup_warmup(state.core.clone(), server_args.clone()));
@@ -91,6 +95,19 @@ pub async fn serve(
     if server_args.metrics.is_some() && metrics_listener {
         router = router.merge(metrics::routes());
     }
+    // Keep this frontend's metrics files current for the directory's other
+    // readers (the node's metrics listener, when this is a DP rank's offset
+    // listener); stops with the runtime.
+    if let Some(frontend) = state.core.metrics().cloned() {
+        tokio::spawn(async move {
+            let mut interval = tokio::time::interval(FLUSH_INTERVAL);
+            loop {
+                interval.tick().await;
+                let frontend = frontend.clone();
+                let _ = tokio::task::spawn_blocking(move || frontend.flush_or_warn()).await;
+            }
+        });
+    }
 
     // TODO(auth): no API-key boundary yet. Python gates every route (except
     // /health*, /metrics*, OPTIONS) via `add_api_key_middleware`; until ported,

rust/sglang-server/src/api_server/http/generate_hook.rs

@@ -0,0 +1,46 @@
+//! The seam a model package uses to add its own keys to native `/generate`
+//! frames — deployment metadata the shared schema does not model. Without a
+//! hook the frames are exactly the `api.v1` schema.
+
+use std::time::Duration;
+
+use serde_json::{Map, Value};
+
+use crate::message::sampling::SamplingParams;
+
+/// Adds keys to native `/generate` frames; installed at launch with
+/// [`crate::Server::start_with_extensions`].
+pub trait GenerateHook: Send + Sync + std::fmt::Debug {
+    /// The annotator for one submitted request, from its sampling parameters
+    /// as submitted (the launch-time preferred params merged in, not yet
+    /// normalized).
+    fn start(&self, sampling_params: &SamplingParams) -> Box<dyn GenerateFrameHook>;
+}
+
+/// One request's annotator, called for each frame the request emits, in
+/// order, the terminal one last.
+pub trait GenerateFrameHook: Send {
+    fn annotate(&mut self, frame: &GenerateFrameInfo, keys: &mut GenerateFrameKeys);
+}
+
+/// What an annotator sees of one frame.
+#[derive(Clone, Copy, Debug, Default)]
+pub struct GenerateFrameInfo {
+    /// The request's terminal frame.
+    pub finished: bool,
+    /// On the terminal frame, the time from the request's first token to its
+    /// last output, as Python's TokenizerManager measures it
+    /// (`get_decode_latency`): for a non-streaming request, from the
+    /// scheduler's prefill stamp when it has one.
+    pub decode_latency: Option<Duration>,
+}
+
+/// Keys an annotator adds to one frame. They must not repeat a key the schema
+/// already writes there.
+#[derive(Debug, Default)]
+pub struct GenerateFrameKeys {
+    /// Added to the frame's `meta_info`.
+    pub meta_info: Map<String, Value>,
+    /// Added to the frame object, after the schema's keys.
+    pub frame: Map<String, Value>,
+}

rust/sglang-server/src/api_server/http/hicache.rs

@@ -99,6 +99,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: (&args).into(),
+                metrics: None,
             },
         );
         (

rust/sglang-server/src/api_server/http/info.rs

@@ -344,6 +344,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: CoreMetadata::from(server_args.as_ref()),
+                metrics: None,
             },
         );
         (
@@ -354,6 +355,7 @@ mod tests {
                 ),
                 server_args,
                 chat_formatter: None,
+                generate_hook: None,
             }),
             intake_rx,
             abort_rx,

rust/sglang-server/src/api_server/http/loads.rs

@@ -12,6 +12,7 @@ use std::fs::File;
 use std::os::unix::fs::FileExt;
 use std::path::Path;
 use std::sync::Arc;
+use std::time::Instant;
 
 use axum::extract::{Query, State};
 use axum::http::{StatusCode, header};
@@ -170,6 +171,7 @@ async fn loads(State(state): State<Arc<AppState>>, Query(query): Query<LoadsQuer
         None | Some(0) => vec![args.load_snapshot_dp_rank],
         Some(_) => Vec::new(),
     };
+    let started = Instant::now();
     let loads: Vec<Map<String, Value>> =
         read_snapshots(Path::new(path), args.load_snapshot_dp_size, &ranks)
             .iter()
@@ -179,6 +181,9 @@ async fn loads(State(state): State<Arc<AppState>>, Query(query): Query<LoadsQuer
                 load
             })
             .collect();
+    if let Some(metrics) = state.core.metrics() {
+        metrics.observe_get_loads(started.elapsed());
+    }
 
     if query.format.as_deref() == Some("prometheus") {
         return (
@@ -474,11 +479,13 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: crate::api_server::core::CoreMetadata::default(),
+                    metrics: None,
                 },
             ),
             generation_policy: crate::message::request::GenerationPolicy::from(&server_args),
             server_args: Arc::new(server_args),
             chat_formatter: None,
+            generate_hook: None,
         });
         let query = Query::try_from_uri(&format!("/v1/loads?{query}").parse().unwrap()).unwrap();
         let response = loads(State(state), query).await;

rust/sglang-server/src/api_server/http/metrics.rs

@@ -1,7 +1,9 @@
 //! `GET /metrics`: the port of Python's `add_prometheus_middleware`. With
 //! `--enable-metrics` every process writes its prometheus_client samples
-//! under `PROMETHEUS_MULTIPROC_DIR`; each scrape merges all of them and
-//! renders the text format (see [`crate::utils::prometheus`]), off the GIL.
+//! under `PROMETHEUS_MULTIPROC_DIR`, this frontend included (see
+//! [`crate::api_server::core::FrontendMetrics`]); each scrape merges all of
+//! them and renders the text format (see [`crate::utils::prometheus`]), off
+//! the GIL.
 
 use std::path::PathBuf;
 use std::sync::Arc;
@@ -33,8 +35,13 @@ async fn metrics(State(state): State<Arc<AppState>>) -> Response {
         return StatusCode::NOT_FOUND.into_response();
     };
     let dir = PathBuf::from(&config.multiproc_dir);
-    // Reading and merging every process's files is blocking file I/O.
+    let frontend = state.core.metrics().cloned();
+    // Writing this frontend's file and reading every process's is blocking
+    // file I/O. The write is best effort: what is on disk is still served.
     let collected = tokio::task::spawn_blocking(move || {
+        if let Some(frontend) = frontend {
+            frontend.flush_or_warn();
+        }
         let files = prometheus::read_dir_files(&dir)?;
         let families = prometheus::merge(
             files
@@ -58,14 +65,15 @@ async fn metrics(State(state): State<Arc<AppState>>) -> Response {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::message::config::{MetricsConfig, ServerArgs};
+    use crate::api_server::core::FrontendMetrics;
+    use crate::message::config::{DisaggregationMode, MetricsConfig, ServerArgs};
     use axum::body::Body;
     use axum::http::Request;
     use tower::ServiceExt;
 
-    fn app(metrics: Option<MetricsConfig>) -> Router {
+    fn app(config: MetricsConfig, frontend: Option<Arc<FrontendMetrics>>) -> Router {
         let server_args = Arc::new(ServerArgs {
-            metrics,
+            metrics: Some(config),
             ..Default::default()
         });
         let state = Arc::new(AppState {
@@ -79,6 +87,7 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: Default::default(),
+                    metrics: frontend,
                 },
             ),
             generation_policy: crate::message::request::GenerationPolicy::from(
@@ -86,6 +95,7 @@ mod tests {
             ),
             server_args,
             chat_formatter: None,
+            generate_hook: None,
         });
         routes().with_state(state)
     }
@@ -111,40 +121,59 @@ mod tests {
         )
     }
 
-    /// A scrape merges every process's file in the directory; a missing
-    /// directory is a 500, as Python's collector raises.
+    /// A scrape merges every process's file in the directory, writing the
+    /// frontend's own first if it can; a missing directory is a 500, as
+    /// Python's collector raises.
     #[tokio::test]
     async fn serves_the_merged_directory() {
         let dir = std::env::temp_dir().join(format!("metrics-route-{}", std::process::id()));
         std::fs::create_dir_all(&dir).unwrap();
         // Two processes' counter files for one series.
-        let file = |value: f64| {
-            let key = br#"["sglang:x", "sglang:x_total", {}, "X."]"#;
-            let mut data = vec![0u8; 8];
-            data.extend_from_slice(&(key.len() as u32).to_le_bytes());
-            data.extend_from_slice(key);
-            data.extend_from_slice(b"    ");
-            data.extend_from_slice(&value.to_le_bytes());
-            data.extend_from_slice(&0f64.to_le_bytes());
-            let used = data.len() as u32;
-            data[..4].copy_from_slice(&used.to_le_bytes());
-            data
-        };
-        std::fs::write(dir.join("counter_1.db"), file(2.0)).unwrap();
-        std::fs::write(dir.join("counter_2.db"), file(3.0)).unwrap();
-        let config = MetricsConfig {
-            multiproc_dir: dir.to_string_lossy().into_owned(),
-        };
+        for (pid, value) in [(1, 2.0), (2, 3.0)] {
+            let entry = prometheus::Entry {
+                key: prometheus::mmap_key("sglang:x", "sglang:x_total", &[], "X."),
+                value,
+                timestamp: 0.0,
+            };
+            let data = prometheus::encode_entries([&entry]);
+            prometheus::write_file(&dir, &format!("counter_{pid}.db"), &data).unwrap();
+        }
+        let config = MetricsConfig::for_dir(&dir);
         let expected =
             "# HELP sglang:x_total X.\n# TYPE sglang:x_total counter\nsglang:x_total 5.0\n";
         for uri in ["/metrics", "/metrics/", "/metrics/any"] {
-            let (status, content_type, body) = get(app(Some(config.clone())), uri).await;
+            let (status, content_type, body) = get(app(config.clone(), None), uri).await;
             assert_eq!(status, StatusCode::OK, "{uri}");
             assert_eq!(content_type, CONTENT_TYPE);
             assert_eq!(body, expected);
         }
+
+        let frontend = Arc::new(FrontendMetrics::new(&config, DisaggregationMode::Null).unwrap());
+        frontend.observe_get_loads(std::time::Duration::from_micros(300));
+        let (_, _, body) = get(app(config.clone(), Some(frontend)), "/metrics").await;
+        assert!(body.contains(expected), "{body}");
+        assert!(
+            body.contains("sglang:get_loads_duration_seconds_bucket{le=\"0.0005\"} 1.0\n"),
+            "{body}"
+        );
+
+        // A frontend file that cannot be written (here a directory in its
+        // place) still serves what is on disk.
+        let blocked = Arc::new(FrontendMetrics::new(&config, DisaggregationMode::Null).unwrap());
+        blocked.observe_get_loads(std::time::Duration::from_micros(300));
+        std::fs::remove_file(dir.join(format!("histogram_rust-{}.db", std::process::id())))
+            .unwrap();
+        std::fs::create_dir_all(
+            dir.join(format!("histogram_rust-{}.db", std::process::id()))
+                .join("x"),
+        )
+        .unwrap();
+        let (status, _, body) = get(app(config.clone(), Some(blocked)), "/metrics").await;
+        assert_eq!(status, StatusCode::OK);
+        assert!(body.contains(expected), "{body}");
+
         std::fs::remove_dir_all(&dir).unwrap();
-        let (status, _, _) = get(app(Some(config)), "/metrics").await;
+        let (status, _, _) = get(app(config, None), "/metrics").await;
         assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
     }
 }

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

@@ -11,6 +11,7 @@ use std::sync::Arc;
 use std::time::{Duration, Instant};
 
 use super::app::AppState;
+use super::generate_hook::{GenerateFrameHook, GenerateFrameInfo, GenerateFrameKeys};
 use super::request_headers::apply_header_overrides;
 use crate::api_server::core::{
     CoreCall, CoreError, CoreEvent, CoreOutput, HealthStatus, recv_indexed,
@@ -75,6 +76,137 @@ impl RequestTiming {
     fn terminal_latencies(&self) -> Option<(Duration, Duration)> {
         Some((self.time_to_first_token?, self.e2e_latency?))
     }
+
+    /// First output to terminal output, once the request finished.
+    fn decode_latency(&self) -> Option<Duration> {
+        let (time_to_first_token, e2e_latency) = self.terminal_latencies()?;
+        Some(e2e_latency.saturating_sub(time_to_first_token))
+    }
+}
+
+/// API-local state for one request: its timing, and the annotator a model
+/// package's [`GenerateHook`](super::generate_hook::GenerateHook) started for
+/// it.
+struct RequestState {
+    timing: RequestTiming,
+    /// The scheduler's decode latency, sent for a finished non-streaming
+    /// request: this frontend sees that request's first output only up to
+    /// `SGLANG_FORCE_STREAM_INTERVAL` tokens late, not when its first token
+    /// was sampled.
+    scheduler_decode_latency: Option<Duration>,
+    hook: Option<Box<dyn GenerateFrameHook>>,
+}
+
+impl RequestState {
+    fn new(state: &AppState, request: &GenerateRequest, timing: RequestTiming) -> Self {
+        Self {
+            timing,
+            scheduler_decode_latency: None,
+            hook: state
+                .generate_hook
+                .as_ref()
+                .map(|hook| hook.start(&request.sampling_params)),
+        }
+    }
+
+    fn finish(&mut self, terminal: &CoreOutput) {
+        self.timing.observe_first_output();
+        self.timing.finish();
+        self.scheduler_decode_latency = terminal.scheduler_decode_latency();
+    }
+
+    /// From the first token to the terminal output, as Python's
+    /// TokenizerManager measures it: the scheduler's measure when it sent one,
+    /// else this frontend's first and terminal outputs.
+    fn decode_latency(&self) -> Option<Duration> {
+        self.scheduler_decode_latency
+            .or_else(|| self.timing.decode_latency())
+    }
+
+    /// One frame as the client receives it: the schema's, plus the hook's keys.
+    fn frame(&mut self, frame: api::GenerateResponse) -> NativeFrame {
+        let finished = frame
+            .meta_info
+            .as_ref()
+            .is_some_and(|meta| meta.finish_reason.is_some());
+        let info = GenerateFrameInfo {
+            finished,
+            decode_latency: if finished {
+                self.decode_latency()
+            } else {
+                None
+            },
+        };
+        let Some(hook) = self.hook.as_mut() else {
+            return NativeFrame { frame, keys: None };
+        };
+        let mut keys = GenerateFrameKeys::default();
+        hook.annotate(&info, &mut keys);
+        NativeFrame {
+            frame,
+            keys: Some(Box::new(keys)),
+        }
+    }
+}
+
+/// One native `/generate` frame: the `api.v1` frame, plus the keys a model
+/// package's hook added after the schema's own.
+#[derive(Debug)]
+struct NativeFrame {
+    frame: api::GenerateResponse,
+    keys: Option<Box<GenerateFrameKeys>>,
+}
+
+impl serde::Serialize for NativeFrame {
+    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
+        use serde::ser::SerializeMap; // codespell:ignore ser
+
+        /// A schema object's keys, then the added ones.
+        #[derive(serde::Serialize)]
+        struct WithKeys<'a, T> {
+            #[serde(flatten)]
+            schema: &'a T,
+            #[serde(flatten)]
+            added: &'a serde_json::Map<String, serde_json::Value>,
+        }
+
+        let Some(keys) = self.keys.as_deref() else {
+            return self.frame.serialize(serializer);
+        };
+        // `GenerateResponse`'s generated serializer, key for key, with the
+        // added keys in `meta_info` and after the frame's own. Destructured
+        // whole so a field the schema gains fails to compile here until it is
+        // written.
+        let api::GenerateResponse {
+            text,
+            meta_info,
+            output_ids,
+            index,
+            prompt_token_ids,
+        } = &self.frame;
+        let mut map = serializer.serialize_map(None)?;
+        map.serialize_entry("text", text)?;
+        if let Some(meta_info) = meta_info {
+            let meta_info = WithKeys {
+                schema: meta_info,
+                added: &keys.meta_info,
+            };
+            map.serialize_entry("meta_info", &meta_info)?;
+        }
+        if let Some(output_ids) = output_ids {
+            map.serialize_entry("output_ids", output_ids)?;
+        }
+        if let Some(index) = index {
+            map.serialize_entry("index", index)?;
+        }
+        if let Some(prompt_token_ids) = prompt_token_ids {
+            map.serialize_entry("prompt_token_ids", prompt_token_ids)?;
+        }
+        for (key, value) in &keys.frame {
+            map.serialize_entry(key, value)?;
+        }
+        map.end()
+    }
 }
 
 /// The routes this module owns, mounted by `api_server::serve`.
@@ -258,6 +390,7 @@ async fn generate_single(
     stream: bool,
     timing: RequestTiming,
 ) -> Response {
+    let request = RequestState::new(state, &req, timing);
     // `return_text_in_logprobs` is decoded on the detok shard into `*_txt`, so
     // the frame builder just reads them — no tokenizer needed here.
     let mut call = match state.core.generate(req).await {
@@ -272,13 +405,13 @@ async fn generate_single(
         // A single request is a 1-element batch without the `index` field — reuse
         // the same stream so the frame/abort/truncation logic lives in one place.
         use futures::StreamExt;
-        let s = generation_event_stream(vec![(call, timing)], incremental, false)
+        let s = generation_event_stream(vec![(call, request)], incremental, false)
             .map(|data| Ok::<_, Infallible>(Event::default().data(data)));
         Sse::new(s).into_response()
     } else {
         // Unary: fold to the terminal and respond once. `CoreCall` disarms
         // itself on a terminal item; truncation leaves it armed for drop cleanup.
-        let (status, item) = drain_unary(&mut call, &rid, timing).await;
+        let (status, item) = drain_unary(&mut call, &rid, request).await;
         (status, Json(item)).into_response()
     }
 }
@@ -290,7 +423,7 @@ async fn generate_single(
 enum GenerateItem {
     // Boxed: a frame is ~300 bytes against the error's ~40, and a batch
     // array holds one per item.
-    Frame(Box<api::GenerateResponse>),
+    Frame(Box<NativeFrame>),
     Error(api::GenerateStreamError),
 }
 
@@ -298,24 +431,24 @@ enum GenerateItem {
 async fn drain_unary(
     call: &mut CoreCall,
     rid_str: &str,
-    mut timing: RequestTiming,
+    mut request: RequestState,
 ) -> (StatusCode, GenerateItem) {
     let mut accumulated = CoreOutput::default();
     while let Some(item) = call.recv().await {
         match item {
             CoreEvent::Delta(out) => {
-                timing.observe_first_output();
+                request.timing.observe_first_output();
                 accumulated.append_delta(&out);
             }
             CoreEvent::Finished(out) => {
-                timing.observe_first_output();
-                timing.finish();
+                request.finish(&out);
                 accumulated.append_delta(&out);
-                let frame = accumulated.frame(rid_str, None, Some(e2e_latency(&timing)));
+                let e2e_latency = Some(e2e_latency(&request.timing));
+                let frame = request.frame(accumulated.frame(rid_str, None, e2e_latency));
                 return (StatusCode::OK, GenerateItem::Frame(Box::new(frame)));
             }
             CoreEvent::Failed(error) => {
-                timing.finish();
+                request.timing.finish();
                 return (
                     error.http_status(),
                     GenerateItem::Error(error.stream_error(None)),
@@ -344,14 +477,15 @@ async fn generate_batch(
     stream: bool,
     timing: RequestTiming,
 ) -> Response {
+    let request_states: Vec<RequestState> = requests
+        .iter()
+        .map(|request| RequestState::new(state, request, timing.clone()))
+        .collect();
     // No cross-item rid collision to worry about: `into_requests` rejected duplicate
     // rids within this batch, and `Rid::from_client` made each one unique against
     // every other in-flight request.
     let calls = match state.core.generate_batch(requests).await {
-        Ok(calls) => calls
-            .into_iter()
-            .map(|call| (call, timing.clone()))
-            .collect(),
+        Ok(calls) => calls.into_iter().zip(request_states).collect(),
         Err(error) => return native_core_error(error, stream),
     };
 
@@ -367,14 +501,13 @@ async fn generate_batch(
         // Unary: poll every item concurrently, as Python's gather does. `join_all`
         // preserves input order for the final JSON array, while each drain observes
         // its own terminal output promptly (important for per-item e2e_latency).
-        let drained = futures::future::join_all(calls.into_iter().map(
-            |(mut call, request_timing)| async move {
+        let drained =
+            futures::future::join_all(calls.into_iter().map(|(mut call, request)| async move {
                 let client_rid = call.public_id().to_owned();
-                let (_status, item) = drain_unary(&mut call, &client_rid, request_timing).await;
+                let (_status, item) = drain_unary(&mut call, &client_rid, request).await;
                 item
-            },
-        ))
-        .await;
+            }))
+            .await;
         (StatusCode::OK, Json(drained)).into_response()
     }
 }
@@ -383,7 +516,7 @@ async fn generate_batch(
 /// each frame (batch only), `incremental` = delta vs cumulative. Each call
 /// aborts itself if the stream is dropped while it is unfinished.
 fn generation_event_stream(
-    calls: Vec<(CoreCall, RequestTiming)>,
+    calls: Vec<(CoreCall, RequestState)>,
     incremental: bool,
     with_index: bool,
 ) -> impl futures::Stream<Item = String> {
@@ -395,10 +528,7 @@ fn generation_event_stream(
             .iter()
             .map(|(call, _)| call.public_id().to_owned())
             .collect();
-        let mut timings: Vec<RequestTiming> = calls
-            .iter()
-            .map(|(_, timing)| timing.clone())
-            .collect();
+        let mut requests: Vec<RequestState> = Vec::with_capacity(n);
         let mut accumulated: Vec<CoreOutput> = (0..n).map(|_| CoreOutput::default()).collect();
 
         // Batch position, tagged onto every frame (a single request omits it).
@@ -407,7 +537,8 @@ fn generation_event_stream(
         // Poll all receivers concurrently; re-arm a receiver's future after each
         // non-terminal frame so its stream keeps flowing.
         let mut futs = futures::stream::FuturesUnordered::new();
-        for (i, (call, _)) in calls.into_iter().enumerate() {
+        for (i, (call, request)) in calls.into_iter().enumerate() {
+            requests.push(request);
             futs.push(recv_indexed(i, call));
         }
 
@@ -429,22 +560,22 @@ fn generation_event_stream(
             for item in items {
                 match item {
                     CoreEvent::Delta(out) => {
-                        timings[i].observe_first_output();
+                        requests[i].timing.observe_first_output();
                         accumulated[i].append_delta(&out);
                         if incremental {
-                            yield json_string(&delta_frame(out, &accumulated[i], &rid_strs[i], idx(i), None));
+                            let frame = delta_frame(out, &accumulated[i], &rid_strs[i], idx(i), None);
+                            yield json_string(&requests[i].frame(frame));
                         } else {
                             coalesced = true;
                         }
                     }
                     CoreEvent::Finished(out) => {
-                        timings[i].observe_first_output();
-                        timings[i].finish();
+                        requests[i].finish(&out);
                         accumulated[i].append_delta(&out);
                         terminal = Some(out);
                     }
                     CoreEvent::Failed(error) => {
-                        timings[i].finish();
+                        requests[i].timing.finish();
                         failed = Some(error);
                     }
                 }
@@ -455,7 +586,7 @@ fn generation_event_stream(
             } else if let Some(out) = terminal {
                 // The final frame carries the full cumulative state, so any
                 // coalesced non-terminal frames are moot.
-                let e2e_latency = Some(e2e_latency(&timings[i]));
+                let e2e_latency = Some(e2e_latency(&requests[i].timing));
                 // The terminal frame is the accumulated output's last use, so
                 // the cumulative form moves it out instead of cloning.
                 let frame = if incremental {
@@ -463,12 +594,13 @@ fn generation_event_stream(
                 } else {
                     std::mem::take(&mut accumulated[i]).frame(&rid_strs[i], idx(i), e2e_latency)
                 };
-                yield json_string(&frame);
+                yield json_string(&requests[i].frame(frame));
             } else {
                 if coalesced {
                     // An intermediate cumulative frame has to clone: the accumulated
                     // output stays behind for the next one.
-                    yield json_string(&accumulated[i].clone().frame(&rid_strs[i], idx(i), None));
+                    let frame = accumulated[i].clone().frame(&rid_strs[i], idx(i), None);
+                    yield json_string(&requests[i].frame(frame));
                 }
                 futs.push(recv_indexed(i, call)); // keep this item flowing
             }
@@ -547,7 +679,7 @@ mod tests {
         serde_json::from_str(s).expect("frame is JSON")
     }
 
-    fn timed_call(rid: u64, rx: mpsc::Receiver<ResponseItem>) -> (CoreCall, RequestTiming) {
+    fn timed_call(rid: u64, rx: mpsc::Receiver<ResponseItem>) -> (CoreCall, RequestState) {
         timed_call_with_abort(rid, rx, flume::unbounded().0)
     }
 
@@ -555,13 +687,17 @@ mod tests {
         rid: u64,
         rx: mpsc::Receiver<ResponseItem>,
         abort_tx: flume::Sender<crate::tokenizer_manager::wiring::AbortSource>,
-    ) -> (CoreCall, RequestTiming) {
+    ) -> (CoreCall, RequestState) {
         (
             CoreCall::from_test_generation_parts(Rid::from(rid.to_string()), rx, abort_tx),
-            RequestTiming {
-                created_at: Instant::now() - Duration::from_millis(10),
-                time_to_first_token: None,
-                e2e_latency: None,
+            RequestState {
+                timing: RequestTiming {
+                    created_at: Instant::now() - Duration::from_millis(10),
+                    time_to_first_token: None,
+                    e2e_latency: None,
+                },
+                scheduler_decode_latency: None,
+                hook: None,
             },
         )
     }
@@ -581,11 +717,13 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: crate::api_server::core::CoreMetadata::default(),
+                    metrics: None,
                 },
             ),
             server_args: Arc::new(crate::message::config::ServerArgs::default()),
             generation_policy: GenerationPolicy::default(),
             chat_formatter: None,
+            generate_hook: None,
         });
 
         let response = health_generate(State(state), Duration::ZERO, true).await;
@@ -621,11 +759,13 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: crate::api_server::core::CoreMetadata::default(),
+                    metrics: None,
                 },
             ),
             generation_policy: crate::message::request::GenerationPolicy::from(&server_args),
             server_args: Arc::new(server_args),
             chat_formatter: None,
+            generate_hook: None,
         });
         // Each probe that reaches intake counts as scheduler output.
         tokio::spawn(async move {
@@ -658,6 +798,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: crate::api_server::core::CoreMetadata::default(),
+                metrics: None,
             },
         );
         let state = Arc::new(AppState {
@@ -665,6 +806,7 @@ mod tests {
             server_args: Arc::new(crate::message::config::ServerArgs::default()),
             generation_policy: GenerationPolicy::default(),
             chat_formatter: None,
+            generate_hook: None,
         });
 
         let responder = tokio::spawn(async move {
@@ -736,6 +878,7 @@ mod tests {
                 is_disaggregation: false,
                 mm_limits: Default::default(),
                 metadata: crate::api_server::core::CoreMetadata::default(),
+                metrics: None,
             },
         );
         let state = Arc::new(AppState {
@@ -743,6 +886,7 @@ mod tests {
             generation_policy: GenerationPolicy::from(&server_args),
             server_args: Arc::new(server_args),
             chat_formatter: None,
+            generate_hook: None,
         });
         let responder = tokio::spawn(async move {
             let mut submitted = Vec::new();
@@ -885,10 +1029,12 @@ mod tests {
             "later output must not overwrite TTFT"
         );
 
+        assert!(timing.decode_latency().is_none(), "unfinished");
         timing.finish();
         let (recorded_ttft, e2e_latency) = timing.terminal_latencies().unwrap();
         assert_eq!(recorded_ttft, time_to_first_token);
         assert!(e2e_latency >= recorded_ttft);
+        assert_eq!(timing.decode_latency(), Some(e2e_latency - recorded_ttft));
 
         timing.finish();
         assert_eq!(
@@ -898,6 +1044,304 @@ mod tests {
         );
     }
 
+    /// A model package's hook: the request's `top_p` on every frame, a
+    /// top-level key on its first frame, and the decode latency on its last.
+    #[derive(Debug)]
+    struct TopPHook;
+
+    struct TopPFrames {
+        top_p: f64,
+        first: bool,
+    }
+
+    impl crate::GenerateHook for TopPHook {
+        fn start(&self, sampling_params: &crate::SamplingParams) -> Box<dyn GenerateFrameHook> {
+            Box::new(TopPFrames {
+                top_p: sampling_params.top_p,
+                first: true,
+            })
+        }
+    }
+
+    impl GenerateFrameHook for TopPFrames {
+        fn annotate(&mut self, frame: &GenerateFrameInfo, keys: &mut GenerateFrameKeys) {
+            assert_eq!(frame.finished, frame.decode_latency.is_some());
+            keys.meta_info
+                .insert("top_p".into(), serde_json::json!(self.top_p));
+            if let Some(latency) = frame.decode_latency {
+                keys.meta_info
+                    .insert("decode_s".into(), serde_json::json!(latency.as_secs_f64()));
+            }
+            if std::mem::take(&mut self.first) {
+                keys.frame.insert("first".into(), serde_json::json!(true));
+            }
+        }
+    }
+
+    /// A hook's keys reach every native frame of every batch item — unary,
+    /// cumulative and incremental streams — each item's hook started from its
+    /// own sampling params, with the scheduler's decode latency for a unary
+    /// item and this frontend's for a streamed one; without a hook the frames
+    /// are the schema's.
+    #[tokio::test]
+    async fn generate_hook_keys_reach_every_frame() {
+        for (stream, incremental, hook) in [
+            (false, false, true),
+            (true, false, true),
+            (true, true, true),
+            (true, true, false),
+        ] {
+            let (intake_tx, intake_rx) = flume::unbounded();
+            let core = crate::api_server::core::CoreHandle::new(
+                intake_tx,
+                flume::unbounded().0,
+                crate::api_server::core::CoreConfig {
+                    response_capacity: 8,
+                    response_activity: Default::default(),
+                    startup_ready: true,
+                    is_disaggregation: false,
+                    mm_limits: Default::default(),
+                    metadata: crate::api_server::core::CoreMetadata::default(),
+                    metrics: None,
+                },
+            );
+            let state = Arc::new(AppState {
+                core,
+                server_args: Arc::new(crate::message::config::ServerArgs {
+                    incremental_streaming_output: incremental,
+                    ..Default::default()
+                }),
+                generation_policy: GenerationPolicy::default(),
+                chat_formatter: None,
+                generate_hook: hook.then(|| Arc::new(TopPHook) as Arc<dyn crate::GenerateHook>),
+            });
+            // Both items' delta, then both terminals: an incremental stream
+            // emits two frames per item.
+            let responder = tokio::spawn(async move {
+                let mut sinks = Vec::new();
+                for _ in 0..2 {
+                    let TmEvent::Intake { request, admission } =
+                        intake_rx.recv_async().await.unwrap()
+                    else {
+                        panic!("expected a generation intake");
+                    };
+                    assert!(admission.try_accept());
+                    sinks.push((request.rid.clone(), request.sink.clone()));
+                }
+                for (rid, sink) in &sinks {
+                    let ResponseItem::Frame(mut chunk) = frame(0, "a") else {
+                        unreachable!()
+                    };
+                    chunk.rid = rid.clone();
+                    sink.try_send(ResponseItem::Frame(chunk)).unwrap();
+                }
+                for (rid, sink) in &sinks {
+                    let ResponseItem::Done(mut chunk) = done(0, "b") else {
+                        unreachable!()
+                    };
+                    chunk.rid = rid.clone();
+                    // The scheduler times non-streaming requests only.
+                    chunk.extras = (!stream).then(|| {
+                        Box::new(crate::message::response::ChunkExtras {
+                            decode_latency: Some(0.25),
+                            ..Default::default()
+                        })
+                    });
+                    sink.try_send(ResponseItem::Done(chunk)).unwrap();
+                }
+            });
+            let body = serde_json::json!({
+                "text": ["x", "y"],
+                "stream": stream,
+                "sampling_params": [{"top_p": 0.5}, {"top_p": 0.25}],
+            });
+            let request = Request::builder()
+                .method("POST")
+                .uri("/generate")
+                .header("content-type", "application/json")
+                .body(Body::from(body.to_string()))
+                .unwrap();
+            let response = routes().with_state(state).oneshot(request).await.unwrap();
+            let body = axum::body::to_bytes(response.into_body(), 64 * 1024)
+                .await
+                .unwrap();
+            responder.await.unwrap();
+            let body = std::str::from_utf8(&body).unwrap();
+            let frames: Vec<serde_json::Value> = if stream {
+                body.lines()
+                    .filter_map(|line| line.strip_prefix("data: "))
+                    .filter(|data| *data != "[DONE]")
+                    .map(parse)
+                    .collect()
+            } else {
+                parse(body).as_array().unwrap().clone()
+            };
+
+            let case = format!("stream={stream} incremental={incremental} hook={hook}");
+            assert_eq!(frames.len(), if incremental { 4 } else { 2 }, "{case}");
+            let mut seen = [0usize; 2];
+            for (position, frame) in frames.iter().enumerate() {
+                // A unary batch is an array in request order, without `index`.
+                let index = frame["index"].as_u64().map_or(position, |i| i as usize);
+                let meta = &frame["meta_info"];
+                let terminal = !meta["finish_reason"].is_null();
+                if !hook {
+                    assert!(meta.get("top_p").is_none() && frame.get("first").is_none());
+                    continue;
+                }
+                assert_eq!(meta["top_p"], [0.5, 0.25][index], "{case}");
+                assert_eq!(frame.get("first").is_some(), seen[index] == 0, "{case}");
+                assert_eq!(meta.get("decode_s").is_some(), terminal, "{case}");
+                if terminal && !stream {
+                    assert_eq!(meta["decode_s"], 0.25, "{case}");
+                }
+                seen[index] += 1;
+            }
+        }
+    }
+
+    /// The hook-aware serializer writes the generated one's keys, in its
+    /// order, before the hook's.
+    #[test]
+    fn native_frame_keeps_the_schema_keys_first() {
+        let schema = api::GenerateResponse {
+            text: "ok".into(),
+            meta_info: Some(api::GenerateMetaInfo {
+                id: "r".into(),
+                prompt_tokens: 2,
+                ..Default::default()
+            }),
+            output_ids: Some(api::TokenIds { ids: vec![7] }),
+            index: Some(1),
+            prompt_token_ids: None,
+        };
+        let expected = json_string(&schema);
+        let empty = NativeFrame {
+            frame: schema.clone(),
+            keys: Some(Box::default()),
+        };
+        assert_eq!(json_string(&empty), expected);
+
+        let mut keys = GenerateFrameKeys::default();
+        keys.meta_info.insert("m".into(), serde_json::json!(1));
+        keys.frame.insert("f".into(), serde_json::json!({"k": "v"}));
+        let added = NativeFrame {
+            frame: schema,
+            keys: Some(Box::new(keys)),
+        };
+        let mut with_meta: serde_json::Value = parse(&expected);
+        with_meta["meta_info"]["m"] = serde_json::json!(1);
+        with_meta["f"] = serde_json::json!({"k": "v"});
+        assert_eq!(json_string(&added), with_meta.to_string());
+    }
+
+    /// A recording hook's annotator: every frame it was handed, in order, each
+    /// frame marked with its position as `meta_info.hook_frame`.
+    #[derive(Clone, Default)]
+    struct Recorded(Arc<std::sync::Mutex<Vec<GenerateFrameInfo>>>);
+
+    impl Recorded {
+        fn hook(&self) -> Option<Box<dyn GenerateFrameHook>> {
+            Some(Box::new(self.clone()))
+        }
+
+        fn frames(&self) -> Vec<GenerateFrameInfo> {
+            self.0.lock().unwrap().clone()
+        }
+    }
+
+    impl GenerateFrameHook for Recorded {
+        fn annotate(&mut self, frame: &GenerateFrameInfo, keys: &mut GenerateFrameKeys) {
+            let mut frames = self.0.lock().unwrap();
+            keys.meta_info
+                .insert("hook_frame".into(), serde_json::json!(frames.len()));
+            frames.push(*frame);
+        }
+    }
+
+    /// A cumulative stream calls the hook once per frame it emits, in order:
+    /// each intermediate frame (one for a coalesced backlog) unfinished and
+    /// without a decode latency, then the terminal one with this frontend's
+    /// decode latency (a streamed request carries no scheduler measure). Each
+    /// frame carries its own call's keys beside its cumulative text.
+    #[tokio::test]
+    async fn generate_hook_annotates_intermediate_cumulative_frames() {
+        let recorded = Recorded::default();
+        let (tx, rx) = mpsc::channel(8);
+        let (call, mut request) = timed_call(10, rx);
+        request.hook = recorded.hook();
+        let stream = generation_event_stream(vec![(call, request)], false, false);
+        futures::pin_mut!(stream);
+
+        tx.send(frame(10, "a")).await.unwrap();
+        let first = parse(&stream.next().await.unwrap());
+        tx.send(frame(10, "b")).await.unwrap();
+        tx.send(frame(10, "c")).await.unwrap();
+        let backlog = parse(&stream.next().await.unwrap());
+        tx.send(done(10, "!")).await.unwrap();
+        let terminal = parse(&stream.next().await.unwrap());
+        assert_eq!(stream.next().await.unwrap(), "[DONE]");
+
+        for (position, (frame, text)) in [(&first, "a"), (&backlog, "abc"), (&terminal, "abc!")]
+            .into_iter()
+            .enumerate()
+        {
+            assert_eq!(frame["text"], text);
+            assert_eq!(frame["meta_info"]["hook_frame"], position, "{text}");
+        }
+        let frames = recorded.frames();
+        let finished: Vec<bool> = frames.iter().map(|frame| frame.finished).collect();
+        assert_eq!(finished, [false, false, true]);
+        assert!(
+            frames[..2]
+                .iter()
+                .all(|frame| frame.decode_latency.is_none())
+        );
+        let decode_latency = frames[2].decode_latency.expect("terminal").as_secs_f64();
+        let e2e_latency = terminal["meta_info"]["e2e_latency"].as_f64().unwrap();
+        assert!(
+            decode_latency <= e2e_latency,
+            "{decode_latency} > {e2e_latency}"
+        );
+    }
+
+    /// The scheduler's decode latency, as Python's `push_generation` frames it
+    /// (`testdata/frame_metadata.json`), reaches the hook on a unary request's
+    /// terminal frame, also from behind logprob columns, in place of this
+    /// frontend's own measure (zero here: the terminal is the first output).
+    #[tokio::test]
+    async fn scheduler_decode_latency_reaches_the_hook() {
+        for (name, rid) in [("batch", "done"), ("logprob_batch", "logprobs")] {
+            let (case, events) = crate::api_server::core::frame_metadata_case(name);
+            let mut event = events
+                .into_iter()
+                .find(|event| event.rid.as_str() == rid)
+                .unwrap();
+            event.rid = Rid::from("10".to_owned());
+            let expected = case["decode_latencies"][rid]
+                .as_f64()
+                .map(Duration::from_secs_f64);
+            assert!(expected.is_some(), "{name}: the fixture times {rid}");
+
+            let recorded = Recorded::default();
+            let (tx, rx) = mpsc::channel(1);
+            let (mut call, mut request) = timed_call(10, rx);
+            request.hook = recorded.hook();
+            tx.send(ResponseItem::Done(event)).await.unwrap();
+            let (status, item) = drain_unary(&mut call, "client-rid", request).await;
+            assert_eq!(status, StatusCode::OK);
+            assert_eq!(
+                serde_json::to_value(&item).unwrap()["meta_info"]["hook_frame"],
+                0
+            );
+
+            let frames = recorded.frames();
+            assert_eq!(frames.len(), 1, "{name}");
+            assert!(frames[0].finished, "{name}");
+            assert_eq!(frames[0].decode_latency, expected, "{name}");
+        }
+    }
+
     /// The native unary response uses the same names and meanings as Python's
     /// TokenizerManager metadata, and adds e2e_latency only on the terminal
     /// result. The timer is seconds from normalized-request acceptance through
@@ -928,7 +1372,12 @@ mod tests {
             time_to_first_token: None,
             e2e_latency: None,
         };
-        let (status, item) = drain_unary(&mut call, "client-rid", timing).await;
+        let request = RequestState {
+            timing,
+            scheduler_decode_latency: None,
+            hook: None,
+        };
+        let (status, item) = drain_unary(&mut call, "client-rid", request).await;
         let value = serde_json::to_value(&item).unwrap();
         assert_eq!(status, StatusCode::OK);
         assert_eq!(value["meta_info"]["id"], "client-rid");

rust/sglang-server/src/api_server/http/openai/chat.rs

@@ -1247,6 +1247,7 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: CoreMetadata::from(args.as_ref()),
+                    metrics: None,
                 },
             );
             let formatter =
@@ -1257,6 +1258,7 @@ mod tests {
                 generation_policy: crate::message::request::GenerationPolicy::from(args.as_ref()),
                 server_args: args,
                 chat_formatter: Some(super::super::ChatFormatter::HuggingFace(formatter)),
+                generate_hook: None,
             });
             let responder = tokio::spawn(async move {
                 let TmEvent::Intake { request, admission } = intake_rx.recv_async().await.unwrap()

rust/sglang-server/src/api_server/http/openai/test_utils.rs

@@ -35,6 +35,7 @@ pub(super) fn core_handle() -> CoreHandle {
             is_disaggregation: false,
             mm_limits: Default::default(),
             metadata: crate::api_server::core::CoreMetadata::from(server_args().as_ref()),
+            metrics: None,
         },
     )
 }
@@ -112,6 +113,7 @@ pub(super) fn app_state(core: CoreHandle) -> Arc<super::AppState> {
         generation_policy: crate::message::request::GenerationPolicy::from(server_args.as_ref()),
         server_args,
         chat_formatter: None,
+        generate_hook: None,
     })
 }
 
@@ -132,6 +134,7 @@ pub(super) fn frontend_closed() -> CoreHandle {
             is_disaggregation: false,
             mm_limits: Default::default(),
             metadata: crate::api_server::core::CoreMetadata::from(server_args().as_ref()),
+            metrics: None,
         },
     )
 }
@@ -157,6 +160,7 @@ fn answering_app_state() -> Arc<super::AppState> {
             is_disaggregation: false,
             mm_limits: Default::default(),
             metadata: crate::api_server::core::CoreMetadata::from(server_args().as_ref()),
+            metrics: None,
         },
     );
     let server_args = server_args();
@@ -169,6 +173,7 @@ fn answering_app_state() -> Arc<super::AppState> {
                 spec: super::template::builtin_template("chatml").unwrap(),
             },
         ))),
+        generate_hook: None,
     })
 }
 

rust/sglang-server/src/api_server/http/openai/tokenize.rs

@@ -271,6 +271,7 @@ mod tests {
                     is_disaggregation: false,
                     mm_limits: Default::default(),
                     metadata: CoreMetadata::from(server_args.as_ref()),
+                    metrics: None,
                 },
             ),
             generation_policy: crate::message::request::GenerationPolicy::from(
@@ -286,6 +287,7 @@ mod tests {
                 )
                 .unwrap(),
             ),
+            generate_hook: None,
         })
     }
 

rust/sglang-server/src/lib.rs

@@ -17,12 +17,16 @@ mod multi_modality;
 mod tokenizer_manager;
 mod utils;
 
+pub use api_server::http::generate_hook::{
+    GenerateFrameHook, GenerateFrameInfo, GenerateFrameKeys, GenerateHook,
+};
 pub use message::config::{
     DefaultSamplingParams, DisaggregationMode, MetricsConfig, MmFamily, MmResample, MmSpec,
     ModelConfig, RustServerServerArgs, ServerArgs,
 };
 pub use message::multimodal::{MediaHints, MmItem};
 pub use message::request::MmData;
+pub use message::sampling::SamplingParams;
 pub use message::types::TokenIds;
 pub use multi_modality::encoded::{
     MRope, MmEncodedEntry, MmEncodedItem, MmMetaValue, MmModality, MmTokenIds,
@@ -152,7 +156,7 @@ fn buffer_to_py(py: Python<'_>, buffer: message::buffers::Buffer) -> PyResult<(S
 }
 
 /// Wraps the HTTP app before it serves: an embedding package's own middleware
-/// or routes ([`Server::start_with_http_layer`]). It wraps every route but the
+/// or routes ([`Server::start_with_extensions`]). It wraps every route but the
 /// PD bootstrap registry's, inside the access log.
 pub type HttpLayer = Arc<dyn Fn(axum::Router) -> axum::Router + Send + Sync>;
 
@@ -188,7 +192,7 @@ impl Server {
         stage_channel_cap: usize,
         cores: Option<Vec<usize>>,
     ) -> PyResult<Self> {
-        Self::start_with_http_layer(
+        Self::start_with_extensions(
             server_args,
             port_offset,
             to_scheduler_cap,
@@ -196,6 +200,7 @@ impl Server {
             stage_channel_cap,
             cores,
             None,
+            None,
         )
     }
 
@@ -288,10 +293,12 @@ impl Server {
 }
 
 impl Server {
-    /// [`Server::start`], with `http_layer` wrapping the HTTP app (see
-    /// [`HttpLayer`]).
+    /// [`Server::start`] with a model package's extensions, the boot of the
+    /// package's own server class: `http_layer` wraps the HTTP app (see
+    /// [`HttpLayer`]), and `generate_hook` adds the package's keys to native
+    /// `/generate` frames (see [`GenerateHook`]).
     #[allow(clippy::too_many_arguments)]
-    pub fn start_with_http_layer(
+    pub fn start_with_extensions(
         server_args: ServerArgs,
         port_offset: Option<u16>,
         to_scheduler_cap: usize,
@@ -299,6 +306,7 @@ impl Server {
         stage_channel_cap: usize,
         cores: Option<Vec<usize>>,
         http_layer: Option<HttpLayer>,
+        generate_hook: Option<Arc<dyn GenerateHook>>,
     ) -> PyResult<Self> {
         // `server_args` already arrived typed (pyo3 rejected any missing/extra/
         // mistyped field when Python constructed it); only value checks remain.
@@ -325,6 +333,7 @@ impl Server {
                 metrics_listener: port_offset.unwrap_or(0) == 0,
             },
             server_args: std::sync::Arc::new(server_args),
+            generate_hook,
         };
         let rt =
             runtime::start(cfg, http_layer).map_err(|e| value_error("runtime start failed", e))?;

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

@@ -68,6 +68,8 @@ pub struct RuntimeConfig {
     /// config-endpoint metadata). `Arc` so cloning the config (and, downstream,
     /// each `AppState`) is cheap; immutable after construction.
     pub server_args: Arc<ServerArgs>,
+    /// A model package's keys for native `/generate` frames.
+    pub generate_hook: Option<Arc<dyn crate::GenerateHook>>,
 }
 
 impl Default for RuntimeConfig {
@@ -75,6 +77,7 @@ impl Default for RuntimeConfig {
         Self {
             rust_server_args: RustServerServerArgs::default(),
             server_args: Arc::new(ServerArgs::default()),
+            generate_hook: None,
         }
     }
 }
@@ -480,16 +483,74 @@ impl<'py> pyo3::FromPyObject<'_, 'py> for DeploymentInfo {
 #[derive(Clone, Debug)]
 pub struct MetricsConfig {
     /// `PROMETHEUS_MULTIPROC_DIR`: every process's prometheus_client files,
-    /// which `/metrics` merges on each scrape.
+    /// which `/metrics` merges on each scrape. The frontend writes its own
+    /// request metrics there too.
     pub multiproc_dir: String,
+    /// The labels on every request metric (`tokenizer_metric_labels()`).
+    pub labels: BTreeMap<String, String>,
+    /// `--enable-priority-scheduling`: the `priority` label carries each
+    /// request's priority.
+    pub priority_label: bool,
+    /// Bucket overrides (`--bucket-*`, resolved `--*-tokens-buckets` rules);
+    /// `None` keeps Python's default buckets.
+    pub bucket_time_to_first_token: Option<Vec<f64>>,
+    pub bucket_inter_token_latency: Option<Vec<f64>>,
+    pub bucket_e2e_request_latency: Option<Vec<f64>>,
+    pub bucket_prompt_tokens: Option<Vec<f64>>,
+    pub bucket_generation_tokens: Option<Vec<f64>>,
 }
 
 #[pyo3::pymethods]
 impl MetricsConfig {
     #[new]
-    #[pyo3(signature = (*, multiproc_dir))]
-    fn py_new(multiproc_dir: String) -> Self {
-        Self { multiproc_dir }
+    #[pyo3(signature = (*,
+        multiproc_dir,
+        labels,
+        priority_label,
+        bucket_time_to_first_token,
+        bucket_inter_token_latency,
+        bucket_e2e_request_latency,
+        bucket_prompt_tokens,
+        bucket_generation_tokens,
+    ))]
+    #[allow(clippy::too_many_arguments)]
+    fn py_new(
+        multiproc_dir: String,
+        labels: BTreeMap<String, String>,
+        priority_label: bool,
+        bucket_time_to_first_token: Option<Vec<f64>>,
+        bucket_inter_token_latency: Option<Vec<f64>>,
+        bucket_e2e_request_latency: Option<Vec<f64>>,
+        bucket_prompt_tokens: Option<Vec<f64>>,
+        bucket_generation_tokens: Option<Vec<f64>>,
+    ) -> Self {
+        Self {
+            multiproc_dir,
+            labels,
+            priority_label,
+            bucket_time_to_first_token,
+            bucket_inter_token_latency,
+            bucket_e2e_request_latency,
+            bucket_prompt_tokens,
+            bucket_generation_tokens,
+        }
+    }
+}
+
+#[cfg(test)]
+impl MetricsConfig {
+    /// Default buckets, no labels, files under `dir`.
+    pub(crate) fn for_dir(dir: &std::path::Path) -> Self {
+        Self {
+            multiproc_dir: dir.to_string_lossy().into_owned(),
+            labels: BTreeMap::new(),
+            priority_label: false,
+            bucket_time_to_first_token: None,
+            bucket_inter_token_latency: None,
+            bucket_e2e_request_latency: None,
+            bucket_prompt_tokens: None,
+            bucket_generation_tokens: None,
+        }
     }
 }
 

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

@@ -47,7 +47,9 @@ pub type ResponseSource = mpsc::Receiver<ResponseItem>;
 /// transport adapter sees them.
 #[derive(Debug)]
 pub enum ResponseItem {
-    /// An intermediate streamed generation step (only sent for streaming reqs).
+    /// An intermediate generation step: every streamed step of a streaming
+    /// request, and every `SGLANG_FORCE_STREAM_INTERVAL` tokens of a
+    /// non-streaming one.
     Frame(ChunkEvent),
     /// The final generation step.
     Done(ChunkEvent),
@@ -187,6 +189,36 @@ pub struct BatchHeader {
     pub hidden_reqlens: Vec<u32>,
     #[serde(default)]
     pub hidden_poslens: Vec<u32>,
+    /// The scheduler's `weight_versions`: a finished request's spans, `None`
+    /// for one still running; empty when no request in the batch finished.
+    #[serde(default)]
+    pub weight_versions: Vec<Option<Vec<WeightVersionSpan>>>,
+    /// Media prompt tokens per modality; empty on a text-only model.
+    #[serde(default)]
+    pub image_tokens: Vec<u32>,
+    #[serde(default)]
+    pub audio_tokens: Vec<u32>,
+    #[serde(default)]
+    pub video_tokens: Vec<u32>,
+    /// A finished non-streaming request's decode latency in seconds, measured
+    /// by the scheduler (`None` for every other row).
+    #[serde(default)]
+    pub decode_latencies: Vec<Option<f64>>,
+    /// Rows the scheduler ended outside the output stream
+    /// (`RustServer.push_abort`), which Python's TokenizerManager answers in
+    /// `_handle_abort_req` without recording request metrics; empty for an
+    /// output-stream batch.
+    #[serde(default)]
+    pub abort_outputs: Vec<bool>,
+}
+
+/// The completion tokens `[start, end)` one weight version produced, as the
+/// scheduler's `WeightVersionSpan` encodes it: `[version, start, end]`.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct WeightVersionSpan {
+    pub version: String,
+    pub start: u64,
+    pub end: u64,
 }
 
 /// Read a request's flat logprob column (`l` val/idx pairs) from `data` at cursors
@@ -303,6 +335,12 @@ pub fn for_each_chunk(body: &[u8], mut route: impl FnMut(ChunkEvent)) -> Decoded
         || !per_req_ok(&h.out_tokids_lp_reqlens)
         || !per_req_ok(&h.in_tokids_lp_reqlens)
         || !per_req_ok(&h.hidden_reqlens)
+        || !(h.weight_versions.is_empty() || h.weight_versions.len() == n)
+        || !per_req_ok(&h.image_tokens)
+        || !per_req_ok(&h.audio_tokens)
+        || !per_req_ok(&h.video_tokens)
+        || !(h.decode_latencies.is_empty() || h.decode_latencies.len() == n)
+        || !(h.abort_outputs.is_empty() || h.abort_outputs.len() == n)
     {
         reject!()
     }
@@ -388,11 +426,40 @@ pub fn for_each_chunk(body: &[u8], mut route: impl FnMut(ChunkEvent)) -> Decoded
     // which is exactly what the streaming decode avoids.
     let mut decode_one = |i: usize| -> Option<ChunkEvent> {
         let token_ids = take_i64(data, &mut c_ids, lens_i(&h.tok_lens, i))?;
+        // The metadata columns after the extras: per-row values, present when a
+        // request in the frame finished or the model is multimodal.
+        let weight_versions = h
+            .weight_versions
+            .get_mut(i)
+            .and_then(Option::take)
+            .unwrap_or_default();
+        let image_tokens = h.image_tokens.get(i).copied().unwrap_or(0);
+        let audio_tokens = h.audio_tokens.get(i).copied().unwrap_or(0);
+        let video_tokens = h.video_tokens.get(i).copied().unwrap_or(0);
+        let decode_latency = h.decode_latencies.get(i).copied().flatten();
+        let abort_output = h.abort_outputs.get(i).copied().unwrap_or(false);
 
         // Plain decode frame (no request in the batch asked for logprobs/hidden):
-        // the extras columns are all zero-width, so skip reading them entirely.
+        // the extras columns are all zero-width, so skip reading them entirely,
+        // and box the metadata only for a row that carries any.
         let extras = if !has_extras {
-            None
+            let has_metadata = !weight_versions.is_empty()
+                || image_tokens != 0
+                || audio_tokens != 0
+                || video_tokens != 0
+                || decode_latency.is_some()
+                || abort_output;
+            has_metadata.then(|| {
+                Box::new(ChunkExtras {
+                    weight_versions,
+                    image_tokens,
+                    audio_tokens,
+                    video_tokens,
+                    decode_latency,
+                    abort_output,
+                    ..Default::default()
+                })
+            })
         } else {
             let (out_lp_val, out_lp_idx) =
                 take_flat(data, &mut c_olp_v, &mut c_olp_i, lens_i(&h.out_lp_lens, i))?;
@@ -459,6 +526,12 @@ pub fn for_each_chunk(body: &[u8], mut route: impl FnMut(ChunkEvent)) -> Decoded
                 in_tid_lens,
                 hidden_val,
                 hidden_lens,
+                weight_versions,
+                image_tokens,
+                audio_tokens,
+                video_tokens,
+                decode_latency,
+                abort_output,
                 // Explicit, NOT `..Default::default()` — same reason as `ChunkEvent`
                 // below: a new column must fail to compile here until it is decoded.
                 out_lp_txt: Vec::new(),
@@ -601,14 +674,16 @@ pub struct ChunkEvent {
     /// Set by the detok shard when the last of `token_ids` is a matched stop
     /// token kept in `output_ids` but left out of `text`.
     pub stop_token_trimmed: bool,
-    /// Logprob + hidden-state columns — `None` unless the request asked for them.
-    /// Boxed to keep the common token/text/finish frame small at large decode
-    /// batches (the decoder allocates it only when a column is non-empty).
+    /// Logprob + hidden-state columns — `None` unless the request asked for them —
+    /// and the rarer per-request metadata. Boxed to keep the common
+    /// token/text/finish frame small at large decode batches (the decoder
+    /// allocates it only when a column is non-empty).
     pub extras: Option<Box<ChunkExtras>>,
 }
 
 /// Logprob + hidden-state columns for a [`ChunkEvent`], allocated only when the
-/// request enabled logprobs / hidden states. Columnar `val`/`idx` (+ ragged `lens`)
+/// request enabled logprobs / hidden states, finished (its weight-version spans)
+/// or has media prompt tokens. Columnar `val`/`idx` (+ ragged `lens`)
 /// buffers arrive pre-decode; the detok shard fills the parallel `*_txt` columns
 /// when `return_text_in_logprobs` is set. In-process only — no serde (see
 /// [`ChunkEvent`]).
@@ -639,6 +714,19 @@ pub struct ChunkExtras {
     /// across chunks (the final message has the full set).
     pub hidden_val: Vec<f32>,
     pub hidden_lens: Vec<u32>,
+    /// Which weight version produced which output tokens; on a finished
+    /// request's chunk only.
+    pub weight_versions: Vec<WeightVersionSpan>,
+    /// Media prompt tokens per modality; zero for a text-only request.
+    pub image_tokens: u32,
+    pub audio_tokens: u32,
+    pub video_tokens: u32,
+    /// A finished non-streaming request's decode latency in seconds, from the
+    /// scheduler.
+    pub decode_latency: Option<f64>,
+    /// The scheduler ended this request outside the output stream (see
+    /// [`BatchHeader::abort_outputs`]).
+    pub abort_output: bool,
     /// Decoded logprob token text (`return_text_in_logprobs`), parallel to the
     /// `*_idx` buffers; empty when not requested (the tuple's text slot stays null).
     pub out_lp_txt: Vec<String>,
@@ -650,8 +738,8 @@ pub struct ChunkExtras {
 }
 
 impl ChunkExtras {
-    /// True when no logprob / hidden column carries data — lets the decoder skip the
-    /// box allocation for the common (extras-free) frame.
+    /// True when no column carries data — lets the decoder skip the box
+    /// allocation for the common (extras-free) frame.
     fn is_empty(&self) -> bool {
         self.out_lp_val.is_empty()
             && self.in_lp_val.is_empty()
@@ -660,6 +748,12 @@ impl ChunkExtras {
             && self.out_tid_lens.is_empty()
             && self.in_tid_lens.is_empty()
             && self.hidden_lens.is_empty()
+            && self.weight_versions.is_empty()
+            && self.image_tokens == 0
+            && self.audio_tokens == 0
+            && self.video_tokens == 0
+            && self.decode_latency.is_none()
+            && !self.abort_output
     }
 }
 
@@ -1240,6 +1334,36 @@ mod tests {
         assert_ne!(events[0].rid, events[1].rid);
     }
 
+    /// A finished request's spans ride the frame without any logprob columns;
+    /// only rows that carry metadata get a `ChunkExtras` box.
+    #[test]
+    fn metadata_columns_box_only_the_rows_that_carry_them() {
+        let span = WeightVersionSpan {
+            version: "v".into(),
+            start: 0,
+            end: 1,
+        };
+        let header = BatchHeader {
+            rids: vec!["a".into(), "b".into(), "c".into()],
+            finish_reasons: vec![None, None, None],
+            prompt_tokens: vec![1, 1, 1],
+            tok_lens: vec![1, 1, 1],
+            cached_tokens: vec![0, 0, 0],
+            reasoning_tokens: vec![0, 0, 0],
+            weight_versions: vec![None, Some(vec![span.clone()]), None],
+            ..Default::default()
+        };
+        let header = rmp_serde::to_vec(&header).unwrap();
+        let data: Vec<u8> = [1i64, 2, 3].iter().flat_map(|x| x.to_le_bytes()).collect();
+        let framed = frame_decode_batch_cols(&header, &[&data]);
+        let mut events = Vec::new();
+        assert!(for_each_chunk(&framed[1..], |ev| events.push(ev)).ok);
+        assert!(events[0].extras.is_none() && events[2].extras.is_none());
+        let extras = events[1].extras.as_deref().unwrap();
+        assert_eq!(extras.weight_versions, [span]);
+        assert!(extras.out_lp_val.is_empty() && extras.image_tokens == 0);
+    }
+
     /// The common frame must stay small: logprob/hidden columns are boxed behind
     /// `ChunkExtras`. 152 since the matched stop token rides in the frame (#42637).
     #[test]

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

@@ -319,6 +319,19 @@ impl SamplingParams {
         self.verify(vocab_size)
     }
 
+    /// `temperature` and `top_k` as the scheduler samples with them, which
+    /// [`normalize`](Self::normalize) settles: a greedy temperature samples at
+    /// 1.0 from the top token, and `top_k = -1` is the whole vocabulary.
+    pub fn sampled_temperature_top_k(&self) -> (f64, i64) {
+        if (0.0..SAMPLING_EPS).contains(&self.temperature) {
+            (1.0, 1)
+        } else if self.top_k == -1 {
+            (self.temperature, TOP_K_ALL)
+        } else {
+            (self.temperature, self.top_k)
+        }
+    }
+
     /// Python `__post_init__` (minus the null-to-default coercions, which the
     /// null-tolerant deserializers above already did): copy the API aliases into
     /// the internal fields and apply the greedy / `top_k` special cases.
@@ -337,14 +350,7 @@ impl SamplingParams {
         if self.stop_token_ids.as_ref().is_some_and(|v| v.is_empty()) {
             self.stop_token_ids = None;
         }
-        if (0.0..SAMPLING_EPS).contains(&self.temperature) {
-            // Greedy: temperature ~0 → temperature=1.0, top_k=1.
-            self.temperature = 1.0;
-            self.top_k = 1;
-        }
-        if self.top_k == -1 {
-            self.top_k = TOP_K_ALL; // -1 disables top_k → whole vocabulary
-        }
+        (self.temperature, self.top_k) = self.sampled_temperature_top_k();
         for constraint in [
             &mut self.json_schema,
             &mut self.regex,

rust/sglang-server/src/utils/prometheus.rs

@@ -5,7 +5,8 @@
 //!
 //! With metrics enabled, every Python process writes its samples to files under
 //! `PROMETHEUS_MULTIPROC_DIR`, and Python's `/metrics` merges those files per
-//! scrape. Reading them here serves the same text without taking the GIL.
+//! scrape. Reading them here serves the same text without taking the GIL, and
+//! the frontend's own samples, written in the same format, merge with them.
 
 use std::collections::{BTreeMap, HashMap};
 use std::fmt::Write as _;
@@ -13,8 +14,8 @@ use std::hash::Hash;
 use std::io;
 use std::path::Path;
 
-/// One `MmapedDict` entry: the JSON key (Python's `mmap_key`), the value and
-/// the timestamp (read only by `mostrecent` gauges).
+/// One `MmapedDict` entry: the JSON key (see [`mmap_key`]), the value and the
+/// timestamp (read only by `mostrecent` gauges).
 #[derive(Clone, Debug, PartialEq)]
 pub struct Entry {
     pub key: String,
@@ -76,6 +77,70 @@ pub fn read_entries(data: &[u8]) -> Result<Vec<Entry>, String> {
     Ok(entries)
 }
 
+/// Encode entries as `MmapedDict` writes them, up to the used length.
+pub fn encode_entries<'a>(entries: impl IntoIterator<Item = &'a Entry>) -> Vec<u8> {
+    let mut out = vec![0u8; HEADER_LEN];
+    for entry in entries {
+        let key = entry.key.as_bytes();
+        out.extend_from_slice(&(key.len() as u32).to_le_bytes());
+        out.extend_from_slice(key);
+        out.resize(out.len() + padded_key_len(key.len()) - key.len(), b' ');
+        out.extend_from_slice(&entry.value.to_le_bytes());
+        out.extend_from_slice(&entry.timestamp.to_le_bytes());
+    }
+    let used = out.len() as u32;
+    out[..4].copy_from_slice(&used.to_le_bytes());
+    out
+}
+
+/// Python `mmap_key`: `json.dumps([metric_name, name, labels, help_text],
+/// sort_keys=True)`, byte for byte.
+pub fn mmap_key(metric_name: &str, name: &str, labels: &[(String, String)], help: &str) -> String {
+    let mut sorted: Vec<&(String, String)> = labels.iter().collect();
+    sorted.sort_by(|a, b| a.0.cmp(&b.0));
+    let mut out = String::from("[");
+    push_json_str(&mut out, metric_name);
+    out.push_str(", ");
+    push_json_str(&mut out, name);
+    out.push_str(", {");
+    for (i, (key, value)) in sorted.into_iter().enumerate() {
+        if i > 0 {
+            out.push_str(", ");
+        }
+        push_json_str(&mut out, key);
+        out.push_str(": ");
+        push_json_str(&mut out, value);
+    }
+    out.push_str("}, ");
+    push_json_str(&mut out, help);
+    out.push(']');
+    out
+}
+
+/// A JSON string as Python's `json.dumps` writes it (`ensure_ascii=True`).
+fn push_json_str(out: &mut String, s: &str) {
+    out.push('"');
+    for c in s.chars() {
+        match c {
+            '"' => out.push_str("\\\""),
+            '\\' => out.push_str("\\\\"),
+            '\n' => out.push_str("\\n"),
+            '\r' => out.push_str("\\r"),
+            '\t' => out.push_str("\\t"),
+            '\u{8}' => out.push_str("\\b"),
+            '\u{c}' => out.push_str("\\f"),
+            ' '..='~' => out.push(c),
+            _ => {
+                let mut units = [0u16; 2];
+                for unit in c.encode_utf16(&mut units) {
+                    let _ = write!(out, "\\u{unit:04x}");
+                }
+            }
+        }
+    }
+    out.push('"');
+}
+
 type Labels = Vec<(String, String)>;
 
 /// One merged sample, labels in the order Python accumulates them.
@@ -503,6 +568,15 @@ pub fn read_dir_files(dir: &Path) -> io::Result<Vec<(String, Vec<u8>)>> {
     Ok(files)
 }
 
+/// Replace `dir/file_name` with `data` (see [`encode_entries`]), atomically: a
+/// scrape reads either the previous file or this one, never a partial write.
+pub fn write_file(dir: &Path, file_name: &str, data: &[u8]) -> io::Result<()> {
+    // Not `*.db`, so a concurrent glob never lists the partial file.
+    let tmp = dir.join(format!("{file_name}.tmp"));
+    std::fs::write(&tmp, data)?;
+    std::fs::rename(&tmp, dir.join(file_name))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -552,6 +626,26 @@ mod tests {
         assert_eq!(render(&families), text);
     }
 
+    /// Re-encoding every fixture file reproduces Python's bytes up to the used
+    /// length (Python preallocates the rest), and each key is what Python's
+    /// `mmap_key` writes for its parsed parts.
+    #[test]
+    fn encodes_files_and_keys_as_python_does() {
+        let (files, _) = fixture();
+        for (name, data) in &files {
+            let entries = read_entries(data).unwrap();
+            let used = u32::from_le_bytes(data[..4].try_into().unwrap()) as usize;
+            assert_eq!(encode_entries(&entries), data[..used], "{name}");
+            for entry in &entries {
+                let (metric_name, sample_name, labels, help) = parse_key(&entry.key).unwrap();
+                assert_eq!(
+                    mmap_key(&metric_name, &sample_name, &labels, &help),
+                    entry.key
+                );
+            }
+        }
+    }
+
     #[test]
     fn corrupt_files_are_skipped() {
         let (files, _) = fixture();
@@ -603,7 +697,8 @@ mod tests {
         let dir = std::env::temp_dir().join(format!("prom-test-{}", std::process::id()));
         std::fs::create_dir_all(&dir).unwrap();
         let (files, _) = fixture();
-        std::fs::write(dir.join(&files[0].0), &files[0].1).unwrap();
+        let entries = read_entries(&files[0].1).unwrap();
+        write_file(&dir, &files[0].0, &encode_entries(&entries)).unwrap();
         std::fs::write(dir.join(".hidden.db"), b"").unwrap();
         std::fs::write(dir.join("counter_1.db.tmp"), b"").unwrap();
         let listed = read_dir_files(&dir).unwrap();

rust/sglang-server/src/utils/runtime.rs

@@ -132,6 +132,15 @@ pub fn start(cfg: RuntimeConfig, http_layer: Option<HttpLayer>) -> Result<Runtim
     let (shutdown_tx, shutdown_rx) = flume::unbounded::<()>();
     let mut threads = Vec::new();
     let plan = plan_cores(&cfg);
+    // Validated before anything is spawned, so a bad bucket list fails startup
+    // cleanly.
+    let metrics = match &cfg.server_args.metrics {
+        Some(config) => Some(Arc::new(crate::api_server::core::FrontendMetrics::new(
+            config,
+            cfg.server_args.disaggregation_mode,
+        )?)),
+        None => None,
+    };
 
     // --- rings (Rust ↔ Python) ---
     let (to_scheduler_tx, to_scheduler_rx): (ToSchedulerTx, ToSchedulerRx) =
@@ -318,6 +327,7 @@ pub fn start(cfg: RuntimeConfig, http_layer: Option<HttpLayer>) -> Result<Runtim
             is_disaggregation: cfg.server_args.is_disaggregation(),
             mm_limits: cfg.server_args.limit_mm_data_per_request.clone(),
             metadata: crate::api_server::core::CoreMetadata::from(cfg.server_args.as_ref()),
+            metrics,
         },
     );
 
@@ -363,6 +373,7 @@ pub fn start(cfg: RuntimeConfig, http_layer: Option<HttpLayer>) -> Result<Runtim
                         cfg.server_args.clone(),
                         cfg.rust_server_args.metrics_listener,
                         http_layer,
+                        cfg.generate_hook.clone(),
                         shutdown_rx.clone(),
                         hicache_manager,
                     );
@@ -423,6 +434,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(test_server_args()),
+            generate_hook: None,
         }
     }
 
@@ -511,6 +523,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(server_args),
+            generate_hook: None,
         };
         // Bind is synchronous in `start`, so the port is already accepting.
         let rt = start(cfg, None).expect("start runtime");
@@ -547,6 +560,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(server_args),
+            generate_hook: None,
         };
         let rt = start(cfg, None).expect("start runtime");
 
@@ -587,9 +601,7 @@ mod tests {
             let mut cfg = test_config(addr, None);
             cfg.rust_server_args.metrics_listener = metrics_listener;
             cfg.server_args = Arc::new(ServerArgs {
-                metrics: Some(crate::message::config::MetricsConfig {
-                    multiproc_dir: dir.to_string_lossy().into_owned(),
-                }),
+                metrics: Some(crate::message::config::MetricsConfig::for_dir(&dir)),
                 ..test_server_args()
             });
             let rt = start(cfg, None).expect("start runtime");
@@ -623,6 +635,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(server_args),
+            generate_hook: None,
         };
         let rt = start(cfg, None).expect("start runtime");
 
@@ -726,6 +739,7 @@ mod tests {
                 ..Default::default()
             },
             server_args: Arc::new(server_args),
+            generate_hook: None,
         };
         let err = match start(cfg, None) {
             Ok(_) => panic!("bind conflict must fail startup, got Ok"),

rust/sglang-server/testdata/frame_metadata.json

@@ -0,0 +1,72 @@
+{
+  "batch": {
+    "header": "dc001792a4646f6e65a772756e6e696e679282a474797065a66c656e677468a66c656e67746803c0920506920301920000920000909090909090909090909090929293a36f6c64000293a36e65770203c092080092000292000092cb3fd0000000000000c0",
+    "data": "0100000000000000020000000000000003000000000000000400000000000000",
+    "decode_latencies": {
+      "done": 0.25
+    },
+    "meta_info": {
+      "done": {
+        "weight_version": "new",
+        "weight_versions": [
+          {
+            "version": "old",
+            "start": 0,
+            "end": 2
+          },
+          {
+            "version": "new",
+            "start": 2,
+            "end": 3
+          }
+        ],
+        "image_tokens": 8
+      },
+      "running": {
+        "weight_version": "launch",
+        "audio_tokens": 2
+      }
+    }
+  },
+  "logprob_batch": {
+    "header": "dc001792a86c6f6770726f6273a5706c61696e9282a474797065a66c656e677468a66c656e67746802c09203049202019200009200009202009090909090909090909090929193a276310002c090909092cb3fe0000000000000c0",
+    "data": "070000000000000008000000000000000900000000000000000000bf000080be0700000008000000",
+    "decode_latencies": {
+      "logprobs": 0.5
+    },
+    "meta_info": {
+      "logprobs": {
+        "weight_version": "v1",
+        "weight_versions": [
+          {
+            "version": "v1",
+            "start": 0,
+            "end": 2
+          }
+        ]
+      },
+      "plain": {
+        "weight_version": "launch"
+      }
+    }
+  },
+  "abort": {
+    "header": "dc001891a761626f727465649182a474797065a561626f7274a76d657373616765b641626f727420696e2077616974696e672071756575659102910091009100909090909090909090909090919293a36f6c64000293a36e657702099090909091c3",
+    "completion_tokens": 3,
+    "meta_info": {
+      "weight_version": "new",
+      "weight_versions": [
+        {
+          "version": "old",
+          "start": 0,
+          "end": 2
+        },
+        {
+          "version": "new",
+          "start": 2,
+          "end": 3
+        }
+      ]
+    }
+  }
+}

rust/sglang-server/testdata/tokenizer_metrics.json

@@ -0,0 +1,395 @@
+{
+  "labels": {
+    "model_name": "model",
+    "engine_type": "unified"
+  },
+  "families": [
+    {
+      "name": "sglang:startup_time_seconds",
+      "kind": "gauge",
+      "documentation": "Engine startup duration by phase in seconds.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "phase"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:startup_cuda_graph_time_seconds",
+      "kind": "gauge",
+      "documentation": "CUDA graph capture duration by phase in seconds.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "phase"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:prompt_tokens_total",
+      "kind": "counter",
+      "documentation": "Number of prefill tokens processed.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:generation_tokens_total",
+      "kind": "counter",
+      "documentation": "Number of generation tokens processed.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:spec_verify_calls_total",
+      "kind": "counter",
+      "documentation": "Number of speculative decoding verification calls.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:prompt_tokens_histogram",
+      "kind": "histogram",
+      "documentation": "Histogram of prompt token length.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": [
+        100.0,
+        300.0,
+        500.0,
+        700.0,
+        1000.0,
+        1500.0,
+        2000.0,
+        3000.0,
+        4000.0,
+        5000.0,
+        6000.0,
+        7000.0,
+        8000.0,
+        9000.0,
+        10000.0,
+        12500.0,
+        15000.0,
+        17500.0,
+        20000.0,
+        22500.0,
+        25000.0,
+        27500.0,
+        30000.0,
+        35000.0,
+        40000.0,
+        60000.0,
+        80000.0,
+        100000.0,
+        200000.0,
+        300000.0,
+        400000.0,
+        600000.0,
+        800000.0,
+        1000000.0,
+        1100000.0
+      ]
+    },
+    {
+      "name": "sglang:uncached_prompt_tokens_histogram",
+      "kind": "histogram",
+      "documentation": "Histogram of uncached (compute) prompt token length.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": [
+        100.0,
+        300.0,
+        500.0,
+        700.0,
+        1000.0,
+        1500.0,
+        2000.0,
+        3000.0,
+        4000.0,
+        5000.0,
+        6000.0,
+        7000.0,
+        8000.0,
+        9000.0,
+        10000.0,
+        12500.0,
+        15000.0,
+        17500.0,
+        20000.0,
+        22500.0,
+        25000.0,
+        27500.0,
+        30000.0,
+        35000.0,
+        40000.0,
+        60000.0,
+        80000.0,
+        100000.0,
+        200000.0,
+        300000.0,
+        400000.0,
+        600000.0,
+        800000.0,
+        1000000.0,
+        1100000.0
+      ]
+    },
+    {
+      "name": "sglang:generation_tokens_histogram",
+      "kind": "histogram",
+      "documentation": "Histogram of generation token length.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": [
+        100.0,
+        300.0,
+        500.0,
+        700.0,
+        1000.0,
+        1500.0,
+        2000.0,
+        3000.0,
+        4000.0,
+        5000.0,
+        6000.0,
+        7000.0,
+        8000.0,
+        9000.0,
+        10000.0,
+        12500.0,
+        15000.0,
+        17500.0,
+        20000.0,
+        22500.0,
+        25000.0,
+        27500.0,
+        30000.0,
+        35000.0,
+        40000.0,
+        60000.0,
+        80000.0,
+        100000.0,
+        200000.0,
+        300000.0,
+        400000.0,
+        600000.0,
+        800000.0,
+        1000000.0,
+        1100000.0
+      ]
+    },
+    {
+      "name": "sglang:cached_tokens_total",
+      "kind": "counter",
+      "documentation": "Number of cached prompt tokens by source (device/host/storage).",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "cache_source"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:num_requests_total",
+      "kind": "counter",
+      "documentation": "Number of requests processed.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:get_loads_duration_seconds",
+      "kind": "histogram",
+      "documentation": "Time spent serving /v1/loads requests (seconds).",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": [
+        0.0001,
+        0.0005,
+        0.001,
+        0.005,
+        0.01,
+        0.05,
+        0.1,
+        0.5,
+        1.0
+      ]
+    },
+    {
+      "name": "sglang:num_so_requests_total",
+      "kind": "counter",
+      "documentation": "Number of structured output requests processed.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:num_aborted_requests_total",
+      "kind": "counter",
+      "documentation": "Number of requests aborted.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": null
+    },
+    {
+      "name": "sglang:time_to_first_token_seconds",
+      "kind": "histogram",
+      "documentation": "Histogram of time to first token in seconds.",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": [
+        0.1,
+        0.2,
+        0.4,
+        0.6,
+        0.8,
+        1.0,
+        2.0,
+        4.0,
+        6.0,
+        8.0,
+        10.0,
+        20.0,
+        40.0,
+        60.0,
+        80.0,
+        100.0,
+        200.0,
+        400.0
+      ]
+    },
+    {
+      "name": "sglang:inter_token_latency_seconds",
+      "kind": "histogram",
+      "documentation": "Histogram of inter-token latency in seconds.",
+      "labelnames": [
+        "model_name",
+        "engine_type"
+      ],
+      "buckets": [
+        0.002,
+        0.004,
+        0.006,
+        0.008,
+        0.01,
+        0.015,
+        0.02,
+        0.025,
+        0.03,
+        0.035,
+        0.04,
+        0.06,
+        0.08,
+        0.1,
+        0.2,
+        0.4,
+        0.6,
+        0.8,
+        1.0,
+        2.0,
+        4.0,
+        6.0,
+        8.0
+      ]
+    },
+    {
+      "name": "sglang:request_time_per_output_token_seconds",
+      "kind": "histogram",
+      "documentation": "Per-request TPOT in seconds: (finished_time - first_token_time) / (completion_tokens - 1).",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": [
+        0.002,
+        0.004,
+        0.006,
+        0.008,
+        0.01,
+        0.015,
+        0.02,
+        0.025,
+        0.03,
+        0.035,
+        0.04,
+        0.06,
+        0.08,
+        0.1,
+        0.2,
+        0.4,
+        0.6,
+        0.8,
+        1.0,
+        2.0,
+        4.0,
+        6.0,
+        8.0
+      ]
+    },
+    {
+      "name": "sglang:e2e_request_latency_seconds",
+      "kind": "histogram",
+      "documentation": "Histogram of End-to-end request latency in seconds",
+      "labelnames": [
+        "model_name",
+        "engine_type",
+        "is_streaming"
+      ],
+      "buckets": [
+        0.1,
+        0.2,
+        0.4,
+        0.6,
+        0.8,
+        1.0,
+        2.0,
+        4.0,
+        6.0,
+        8.0,
+        10.0,
+        20.0,
+        40.0,
+        60.0,
+        80.0,
+        100.0,
+        200.0,
+        400.0,
+        600.0,
+        1200.0,
+        1800.0,
+        2400.0
+      ]
+    }
+  ]
+}

test/registered/unit/api/v1/test_parity.py

@@ -288,6 +288,9 @@ class TestResponseParity(CustomTestCase):
                 "output_token_logprobs_length",
                 "weight_version",
                 "weight_versions",
+                "image_tokens",
+                "audio_tokens",
+                "video_tokens",
             ],
         )
 

test/registered/unit/entrypoints/test_rust_server_dp_ports.py

@@ -160,10 +160,17 @@ def test_deployment_info_reports_listener_dp_rank(server_args, ranks, expected):
 
 
 @pytest.mark.parametrize("enable_metrics", [False, True])
-def test_typed_config_points_metrics_at_the_multiproc_dir(enable_metrics):
+def test_typed_config_carries_tokenizer_metrics(enable_metrics):
     extension = MagicMock()
     with (
-        get_context().override_server_args(enable_metrics=enable_metrics),
+        get_context().override_server_args(
+            enable_metrics=enable_metrics,
+            served_model_name="m",
+            enable_priority_scheduling=True,
+            extra_metric_labels={"cluster": "c"},
+            bucket_inter_token_latency=[0.001, 0.01],
+            generation_tokens_buckets=["custom", "10", "1"],
+        ),
         get_parallel().override(attn_dp_rank=0, dp_rank=None),
         patch.dict("os.environ", {"PROMETHEUS_MULTIPROC_DIR": "/tmp/prom"}),
         patch.object(rust_extensions, "load_rust_extension", return_value=extension),
@@ -175,11 +182,26 @@ def test_typed_config_points_metrics_at_the_multiproc_dir(enable_metrics):
         )
 
     metrics = extension.ServerArgs.call_args.kwargs["metrics"]
-    if enable_metrics:
-        extension.MetricsConfig.assert_called_once_with(multiproc_dir="/tmp/prom")
-        assert metrics is extension.MetricsConfig.return_value
-    else:
+    if not enable_metrics:
         assert metrics is None
+        return
+    assert metrics is extension.MetricsConfig.return_value
+    extension.MetricsConfig.assert_called_once_with(
+        multiproc_dir="/tmp/prom",
+        labels={
+            "model_name": "m",
+            "engine_type": "unified",
+            "priority": "",
+            "cluster": "c",
+        },
+        priority_label=True,
+        bucket_time_to_first_token=None,
+        bucket_inter_token_latency=[0.001, 0.01],
+        bucket_e2e_request_latency=None,
+        # The default rule keeps the Rust side's defaults; others resolve here.
+        bucket_prompt_tokens=None,
+        bucket_generation_tokens=[1.0, 10.0],
+    )
 
 
 @pytest.mark.parametrize(

test/registered/unit/managers/test_output_streamer_customized_info.py

@@ -1,5 +1,6 @@
 import dataclasses
 import pickle
+import time
 import unittest
 from array import array
 from types import SimpleNamespace
@@ -105,6 +106,7 @@ def _accumulator(
     current_weight_version="default",
     return_sampling_mask=False,
     request_level_keys=frozenset(),
+    rust_server_mode=False,
 ):
     return _GenerationStreamAccumulator(
         return_logprob=False,
@@ -119,24 +121,30 @@ def _accumulator(
         get_cached_tokens_details=lambda req: None,
         is_request_level_customized_info_key=request_level_keys.__contains__,
         current_weight_version=current_weight_version,
+        rust_server_mode=rust_server_mode,
     )
 
 
+def _enter_streamer_scope(test):
+    """Patch what `SchedulerOutputStreamer` reads per call, for `test`."""
+    serving_patch = patch(
+        "sglang.srt.managers.scheduler_components.output_streamer.get_serving",
+        return_value=SimpleNamespace(stream_interval=1, weight_version="default"),
+    )
+    observability_patch = patch(
+        "sglang.srt.managers.scheduler_components.output_streamer.get_observability",
+        return_value=SimpleNamespace(enable_request_time_stats_logging=False),
+    )
+    serving_patch.start()
+    observability_patch.start()
+    enter_scope(test, published_topology(ranks={"dp_rank": 0}))
+    test.addCleanup(serving_patch.stop)
+    test.addCleanup(observability_patch.stop)
+
+
 class TestOutputStreamerCustomizedInfo(unittest.TestCase):
     def setUp(self):
-        serving_patch = patch(
-            "sglang.srt.managers.scheduler_components.output_streamer.get_serving",
-            return_value=SimpleNamespace(stream_interval=1, weight_version="default"),
-        )
-        observability_patch = patch(
-            "sglang.srt.managers.scheduler_components.output_streamer.get_observability",
-            return_value=SimpleNamespace(enable_request_time_stats_logging=False),
-        )
-        serving_patch.start()
-        observability_patch.start()
-        enter_scope(self, published_topology(ranks={"dp_rank": 0}))
-        self.addCleanup(serving_patch.stop)
-        self.addCleanup(observability_patch.stop)
+        _enter_streamer_scope(self)
 
     def test_customized_info_is_padded_for_mixed_batches(self):
         accumulator = _accumulator()
@@ -501,6 +509,56 @@ class TestOutputStreamerWeightVersions(unittest.TestCase):
         self.assertIsNone(payload.weight_versions)
 
 
+class TestOutputStreamerDecodeLatency(unittest.TestCase):
+    def setUp(self):
+        _enter_streamer_scope(self)
+
+    def test_rust_mode_times_finished_non_streaming_requests(self):
+        """In Rust mode a finished non-streaming request's decode latency runs
+        from the prefill stamp, which Python's TokenizerManager reads only with
+        metrics on, and the scheduler hands it to the Rust frontend by payload
+        row with the batch; other requests, and Python mode, record none."""
+        stamp = time.perf_counter() - 0.5
+
+        def req(rid, *, stream=False, finished=True, metrics=True, stamp=stamp):
+            req = _FakeReq(rid, [1], finished=finished)
+            req.stream = stream
+            req.time_stats = SimpleNamespace(
+                enable_metrics=metrics, prefill_finished_time=stamp
+            )
+            return req
+
+        rust_server = Mock()
+        streamer = SchedulerOutputStreamer(
+            send_to_detokenizer=None,
+            tree_cache=None,
+            server_args=SimpleNamespace(),
+            is_generation=True,
+            spec_algorithm=SpeculativeAlgorithm.NONE,
+            disaggregation_mode=DisaggregationMode.NULL,
+            enable_hicache_storage=lambda: False,
+            rust_server=rust_server,
+        )
+        streamer._stream_output_generation(
+            [
+                req("streaming", stream=True),
+                req("timed"),
+                req("running", finished=False),
+                req("no_metrics", metrics=False),
+                req("no_stamp", stamp=0.0),
+            ],
+            False,
+        )
+        payload, decode_latencies = rust_server.push_generation.call_args.args
+        self.assertEqual(payload.rids[1], "timed")
+        self.assertEqual(list(decode_latencies), [1])
+        self.assertGreaterEqual(decode_latencies[1], 0.5)
+
+        python_mode = _accumulator()
+        python_mode.accept(req=req("timed"))
+        self.assertEqual(python_mode.decode_latencies, {})
+
+
 class TestOutputStreamerStreamInterval(unittest.TestCase):
     def test_sends_once_interval_tokens_are_unsent(self):
         """With stream_interval=4, the first token goes out at once, then a chunk

test/registered/unit/multimodal/rust/shared/test_rust_server_extension.py

@@ -1,6 +1,9 @@
 """Model packages use one extension for server arguments and worker startup."""
 
+import json
+import os
 import unittest
+from array import array
 from contextlib import contextmanager
 from types import ModuleType, SimpleNamespace
 from unittest.mock import Mock, patch, sentinel
@@ -15,6 +18,137 @@ from sglang.srt.rust_server.server import RustServer  # noqa: E402
 
 register_cpu_ci(est_time=1, suite="base-a-test-cpu")
 
+# The Rust frontend decodes these headers and renders their `meta_info` keys
+# (rust/sglang-server/src/api_server/core/frame.rs); they must match Python.
+RUST_FRAME_METADATA_FIXTURE = os.path.join(
+    os.path.dirname(__file__),
+    *[os.pardir] * 6,
+    "rust/sglang-server/testdata/frame_metadata.json",
+)
+
+
+def build_rust_frame_metadata_fixture() -> dict:
+    """The headers `push_generation` and `push_abort` send for the weight-version,
+    media-count and decode-latency columns, and the `meta_info` keys Python's
+    TokenizerManager derives from the same values, with "launch" as the
+    server's weight version.
+    Regenerate the fixture with ``json.dump(build_rust_frame_metadata_fixture(),
+    f, indent=2)`` after changing either side."""
+    from sglang.srt.managers.io_struct import AbortReq
+    from sglang.srt.utils.weight_versions import (
+        WeightVersionSpan,
+        add_weight_versions_to_meta_info,
+    )
+
+    def meta_info(spans, completion_tokens, image=0, audio=0, video=0):
+        # TokenizerManager._handle_batch_output / _handle_abort_req.
+        meta = {"weight_version": "launch"}
+        if spans is not None:
+            add_weight_versions_to_meta_info(meta, spans, completion_tokens)
+        for key, count in (
+            ("image_tokens", image),
+            ("audio_tokens", audio),
+            ("video_tokens", video),
+        ):
+            if count:
+                meta[key] = count
+        return meta
+
+    no_extras = dict.fromkeys(
+        [
+            f"{family}_{part}"
+            for family in (
+                "output_token_logprobs",
+                "input_token_logprobs",
+                "output_top_logprobs",
+                "input_top_logprobs",
+                "output_token_ids_logprobs",
+                "input_token_ids_logprobs",
+            )
+            for part in ("val", "idx")
+        ]
+        + ["output_hidden_states"]
+    )
+
+    def push_generation(multimodal, decode_latencies, **columns):
+        server = RustServer(Mock(), http_port=0)
+        server._multimodal_enabled = multimodal
+        payload = SimpleNamespace(**{**no_extras, **columns})
+        server.push_generation(payload, decode_latencies)
+        header, data = server.server.push_decode_result_batch.call_args.args
+        return {
+            "header": header.hex(),
+            "data": b"".join(data).hex(),
+            "decode_latencies": {
+                columns["rids"][row]: latency
+                for row, latency in decode_latencies.items()
+            },
+        }
+
+    swapped = [
+        WeightVersionSpan(version="old", start=0, end=2),
+        WeightVersionSpan(version="new", start=2, end=3),
+    ]
+    batch = push_generation(
+        True,
+        {0: 0.25},
+        rids=["done", "running"],
+        finished_reasons=[{"type": "length", "length": 3}, None],
+        prompt_tokens=[5, 6],
+        output_ids=[array("q", [1, 2, 3]), array("q", [4])],
+        cached_tokens=[0, 0],
+        reasoning_tokens=[0, 0],
+        weight_versions=[swapped, None],
+        image_tokens=[8, 0],
+        audio_tokens=[0, 2],
+        video_tokens=[0, 0],
+    )
+    batch["meta_info"] = {
+        "done": meta_info(swapped, 3, image=8),
+        "running": meta_info(None, 1, audio=2),
+    }
+    # A text model's frame with output logprobs: the spans and the decode
+    # latency follow the extras columns rather than their padding.
+    single = [WeightVersionSpan(version="v1", start=0, end=2)]
+    logprob_batch = push_generation(
+        False,
+        {0: 0.5},
+        rids=["logprobs", "plain"],
+        finished_reasons=[{"type": "length", "length": 2}, None],
+        prompt_tokens=[3, 4],
+        output_ids=[array("q", [7, 8]), array("q", [9])],
+        cached_tokens=[0, 0],
+        reasoning_tokens=[0, 0],
+        weight_versions=[single, None],
+        image_tokens=[0, 0],
+        audio_tokens=[0, 0],
+        video_tokens=[0, 0],
+        output_token_logprobs_val=[[-0.5, -0.25], []],
+        output_token_logprobs_idx=[[7, 8], []],
+    )
+    logprob_batch["meta_info"] = {
+        "logprobs": meta_info(single, 2),
+        "plain": meta_info(None, 1),
+    }
+    # The scheduler's spans run past the three tokens the client received.
+    aborted = [
+        WeightVersionSpan(version="old", start=0, end=2),
+        WeightVersionSpan(version="new", start=2, end=9),
+    ]
+    server = RustServer(Mock(), http_port=0)
+    req = SimpleNamespace(origin_input_ids=[1, 2], cached_tokens=0, reasoning_tokens=0)
+    server.push_abort(AbortReq(rid="aborted", weight_versions=aborted), req)
+    abort_header, _ = server.server.push_decode_result_batch.call_args.args
+    return {
+        "batch": batch,
+        "logprob_batch": logprob_batch,
+        "abort": {
+            "header": abort_header.hex(),
+            "completion_tokens": 3,
+            "meta_info": meta_info(aborted, 3),
+        },
+    }
+
 
 @contextmanager
 def launch_context(server_cls):
@@ -68,6 +202,10 @@ class TestRustServerExtension(CustomTestCase):
             def _load_extension(cls):
                 return extension
 
+            @classmethod
+            def _server_kwargs(cls, scheduler):
+                return {"model_setting": sentinel.setting}
+
             @classmethod
             def _resolve_multimodal(cls, scheduler):
                 return None
@@ -88,7 +226,10 @@ class TestRustServerExtension(CustomTestCase):
             default_max_new_tokens=128,
         )
         extension.Server.assert_called_once_with(
-            sentinel.args, cores=None, port_offset=1
+            sentinel.args,
+            cores=None,
+            port_offset=1,
+            model_setting=sentinel.setting,
         )
         instance.server.start_mm_workers.assert_called_once_with(sentinel.spec, 8)
         self.assertIsInstance(instance, ModelServer)
@@ -122,8 +263,9 @@ class TestRustServerExtension(CustomTestCase):
 
     def test_scheduler_aborts_reach_the_rust_egress(self):
         """An abort the scheduler sends outside the output stream becomes a
-        one-request terminal batch in Rust mode, keeping the request's counts;
-        without the Rust server it still goes to the tokenizer socket."""
+        one-request terminal batch in Rust mode, keeping the request's counts
+        and marked as outside the stream; without the Rust server it still
+        goes to the tokenizer socket."""
         import msgspec
 
         from sglang.srt.managers.io_struct import AbortReq
@@ -139,7 +281,9 @@ class TestRustServerExtension(CustomTestCase):
         header, cols = server.server.push_decode_result_batch.call_args.args
         self.assertEqual(
             msgspec.msgpack.decode(header),
-            [["r#1"], [{"type": "abort", "message": "gone"}], [3], [0], [2], [1]],
+            [["r#1"], [{"type": "abort", "message": "gone"}], [3], [0], [2], [1]]
+            + [[]] * 17
+            + [[True]],
         )
         self.assertEqual(cols, [b""])
         scheduler.ipc_channels.send_to_tokenizer.send_output.assert_not_called()
@@ -150,6 +294,10 @@ class TestRustServerExtension(CustomTestCase):
             abort, req
         )
 
+    def test_frame_metadata_fixture_matches_python(self):
+        with open(RUST_FRAME_METADATA_FIXTURE) as f:
+            self.assertEqual(json.load(f), build_rust_frame_metadata_fixture())
+
     def test_launch_resolves_the_mm_pipeline_before_the_server_starts(self):
         """The listener answers once the server is constructed, so a model
         without a Rust MM pipeline must be refused before that; a supported

test/registered/unit/observability/test_rust_metrics_fixtures.py

@@ -1,10 +1,11 @@
 """Fixtures pinning the Rust frontend's ``/metrics`` to Python's.
 
 The Rust server (``rust/sglang-server``) reads prometheus_client's multiprocess
-files and renders them itself (``src/utils/prometheus.rs``). These tests fail if
-prometheus_client's file format, merge or text rendering drifts from what the
-committed fixtures record; regenerate them as each builder's docstring says,
-then update the Rust side.
+files and renders them itself (``src/utils/prometheus.rs``), and records the
+``TokenizerMetricsCollector`` families itself (``src/api_server/core/metrics.rs``).
+These tests fail if prometheus_client's file format, merge or text rendering, or
+the collector's families, drift from what the committed fixtures record;
+regenerate them as each builder's docstring says, then update the Rust side.
 """
 
 from sglang.test.ci.ci_register import register_cpu_ci
@@ -22,12 +23,15 @@ from prometheus_client import Counter, Gauge, Histogram, Summary, values
 from prometheus_client.exposition import generate_latest
 from prometheus_client.multiprocess import MultiProcessCollector
 
+from sglang.srt.observability.metrics_collector import TokenizerMetricsCollector
+from sglang.srt.runtime_context import get_context
 from sglang.test.test_utils import CustomTestCase
 
 TESTDATA = os.path.join(
     os.path.dirname(__file__), *[os.pardir] * 4, "rust/sglang-server/testdata"
 )
 MULTIPROCESS_FIXTURE = os.path.join(TESTDATA, "prometheus_multiprocess.json")
+TOKENIZER_FIXTURE = os.path.join(TESTDATA, "tokenizer_metrics.json")
 
 
 def _write_process(directory: str, pid: int, now: float, observe) -> None:
@@ -135,7 +139,46 @@ def _render_python(files: list) -> str:
         return generate_latest(_Files()).decode()
 
 
+def build_rust_tokenizer_metrics_fixture() -> dict:
+    """Every family ``TokenizerMetricsCollector`` registers under the default
+    server args, as its constructor arguments. Regenerate the fixture with
+    ``json.dump(build_rust_tokenizer_metrics_fixture(), f, indent=2)``."""
+    families = []
+
+    def recorder(kind):
+        def register(name, documentation, labelnames=(), buckets=None, **_):
+            families.append(
+                {
+                    "name": name,
+                    "kind": kind,
+                    "documentation": documentation,
+                    "labelnames": list(labelnames),
+                    "buckets": (
+                        None if buckets is None else [float(b) for b in buckets]
+                    ),
+                }
+            )
+            return mock.MagicMock()
+
+        return staticmethod(register)
+
+    class _Recording(TokenizerMetricsCollector):
+        _counter_cls = recorder("counter")
+        _gauge_cls = recorder("gauge")
+        _histogram_cls = recorder("histogram")
+
+    labels = {"model_name": "model", "engine_type": "unified"}
+    with get_context().override_server_args():
+        _Recording(labels=labels)
+    return {"labels": labels, "families": families}
+
+
 class TestRustMetricsFixtures(CustomTestCase):
+    def test_tokenizer_metrics_fixture_matches_python(self):
+        with open(TOKENIZER_FIXTURE) as f:
+            fixture = json.load(f)
+        self.assertEqual(fixture, build_rust_tokenizer_metrics_fixture())
+
     def test_multiprocess_fixture_matches_python(self):
         with open(MULTIPROCESS_FIXTURE) as f:
             fixture = json.load(f)