feat: expose scheduler fresh prefill counter

tests/v1/core/test_scheduler.py

@@ -100,6 +100,7 @@ def test_schedule(enable_prefix_caching: bool, prompt_logprobs: int | None):
     # Verify all requests are scheduled.
     for req_id, num_tokens in output.num_scheduled_tokens.items():
         assert num_tokens == len(requests[int(req_id)].prompt_token_ids)
+    assert output.prefill_tokens_scheduled == output.total_num_scheduled_tokens
 
     # Verify requests moved from waiting to running
     assert len(scheduler.waiting) == 0
@@ -867,6 +868,12 @@ def test_schedule_concurrent_batches(
     # The first request is still running, so only schedule the second request.
     scheduler.add_request(requests[1])
     scheduler_output1 = scheduler.schedule()
+
+    # Initial scheduling is all fresh prompt work, including when the prompt
+    # is split across multiple chunked-prefill steps.
+    assert scheduler_output1.prefill_tokens_scheduled == (
+        scheduler_output1.total_num_scheduled_tokens
+    )
     assert len(scheduler_output1.scheduled_new_reqs) == 1
     assert scheduler_output1.num_scheduled_tokens[requests[1].request_id] == 512
 

vllm/v1/core/sched/output.py

@@ -194,6 +194,9 @@ class SchedulerOutput:
     # Total number of tokens scheduled for all requests.
     # Equal to sum(num_scheduled_tokens.values())
     total_num_scheduled_tokens: int
+    # Fresh prompt tokens issued in this scheduler step. Prefix-cache and
+    # external-KV tokens are excluded.
+    prefill_tokens_scheduled: int
     # req_id -> spec_token_ids
     # If a request does not have any spec decode tokens, it will not be
     # included in the dictionary.
@@ -251,6 +254,7 @@ def make_empty(cls) -> "SchedulerOutput":
             scheduled_cached_reqs=CachedRequestData.make_empty(),
             num_scheduled_tokens={},
             total_num_scheduled_tokens=0,
+            prefill_tokens_scheduled=0,
             scheduled_spec_decode_tokens={},
             scheduled_encoder_inputs={},
             num_common_prefix_blocks=[],

vllm/v1/core/sched/scheduler.py

@@ -425,6 +425,7 @@ def schedule(self, throttle_prefills: bool = False) -> SchedulerOutput:
         scheduled_spec_decode_tokens: dict[str, list[int]] = {}
         # Whether the running batch contains any prefill requests.
         prefill_scheduled = False
+        prefill_tokens_scheduled = 0
 
         # For logging.
         scheduled_timestamp = time.monotonic()
@@ -584,6 +585,13 @@ def schedule(self, throttle_prefills: bool = False) -> SchedulerOutput:
             # Schedule the request.
             scheduled_running_reqs.append(request)
             prefill_scheduled |= request.is_prefill_chunk
+            prefill_tokens_scheduled += max(
+                0,
+                min(
+                    num_new_tokens,
+                    request.num_prompt_tokens - request.num_computed_tokens,
+                ),
+            )
             request_id = request.request_id
             req_to_new_blocks[request_id] = new_blocks
             num_scheduled_tokens[request_id] = num_new_tokens
@@ -986,6 +994,13 @@ def schedule(self, throttle_prefills: bool = False) -> SchedulerOutput:
                     request_id
                 )
                 num_scheduled_tokens[request_id] = num_new_tokens
+                prefill_tokens_scheduled += max(
+                    0,
+                    min(
+                        num_new_tokens,
+                        request.num_prompt_tokens - num_computed_tokens,
+                    ),
+                )
                 token_budget -= num_new_tokens
                 request.status = RequestStatus.RUNNING
                 request.num_computed_tokens = num_computed_tokens
@@ -1097,6 +1112,7 @@ def schedule(self, throttle_prefills: bool = False) -> SchedulerOutput:
             scheduled_cached_reqs=cached_reqs_data,
             num_scheduled_tokens=num_scheduled_tokens,
             total_num_scheduled_tokens=total_num_scheduled_tokens,
+            prefill_tokens_scheduled=prefill_tokens_scheduled,
             scheduled_spec_decode_tokens=scheduled_spec_decode_tokens,
             scheduled_encoder_inputs=scheduled_encoder_inputs,
             num_common_prefix_blocks=num_common_prefix_blocks,
@@ -1844,7 +1860,11 @@ def update_from_output(
 
         if (
             stats := self.make_stats(
-                spec_decoding_stats, kv_connector_stats, cudagraph_stats, perf_stats
+                spec_decoding_stats,
+                kv_connector_stats,
+                cudagraph_stats,
+                perf_stats,
+                scheduler_output.prefill_tokens_scheduled,
             )
         ) is not None:
             # Return stats to only one of the front-ends.
@@ -2285,6 +2305,7 @@ def make_stats(
         kv_connector_stats: KVConnectorStats | None = None,
         cudagraph_stats: CUDAGraphStat | None = None,
         perf_stats: PerfStats | None = None,
+        prefill_tokens_scheduled: int = 0,
     ) -> SchedulerStats | None:
         if not self.log_stats:
             return None
@@ -2307,6 +2328,7 @@ def make_stats(
             num_running_reqs=len(self.running),
             num_waiting_reqs=len(self.waiting),
             num_skipped_waiting_reqs=len(self.skipped_waiting),
+            prefill_tokens_scheduled=prefill_tokens_scheduled,
             kv_cache_usage=self.kv_cache_manager.usage,
             prefix_cache_stats=prefix_cache_stats,
             connector_prefix_cache_stats=connector_prefix_cache_stats,

vllm/v1/engine/llm_engine.py

@@ -322,7 +322,6 @@ def step(self) -> list[RequestOutput | PoolingRequestOutput]:
             if (
                 self.logger_manager is not None
                 and outputs.scheduler_stats is not None
-                and len(outputs.outputs) > 0
             ):
                 self.logger_manager.record(
                     scheduler_stats=outputs.scheduler_stats,

vllm/v1/metrics/loggers.py

@@ -639,6 +639,18 @@ def __init__(
             counter_prompt_tokens, per_engine_labelvalues
         )
 
+        counter_prefill_tokens_scheduled = self._counter_cls(
+            name="vllm:prefill_tokens_scheduled",
+            documentation=(
+                "Fresh prompt tokens issued by the scheduler for local prefill. "
+                "Excludes prefix-cache and external-KV tokens."
+            ),
+            labelnames=labelnames,
+        )
+        self.counter_prefill_tokens_scheduled = create_metric_per_engine(
+            counter_prefill_tokens_scheduled, per_engine_labelvalues
+        )
+
         # Labeled prompt token counters by source
         counter_prompt_tokens_by_source = self._counter_cls(
             name="vllm:prompt_tokens_by_source",
@@ -1069,6 +1081,9 @@ def record(
     ):
         """Log to prometheus."""
         if scheduler_stats is not None:
+            self.counter_prefill_tokens_scheduled[engine_idx].inc(
+                scheduler_stats.prefill_tokens_scheduled
+            )
             self.gauge_scheduler_running[engine_idx].set(
                 scheduler_stats.num_running_reqs
             )

vllm/v1/metrics/stats.py

@@ -176,6 +176,12 @@ class SchedulerStats:
     num_waiting_reqs: int = 0  # length of the "waiting" request queue
     num_skipped_waiting_reqs: int = 0  # length of the "skipped waiting" queue
 
+    # Fresh prompt tokens issued to the model in the current scheduler step.
+    # This excludes prefix-cache and external-KV tokens, and is intentionally
+    # separate from request-level prompt accounting, which may be reported only
+    # when a request produces its first output.
+    prefill_tokens_scheduled: int = 0
+
     # These are used for internal DP load-balancing.
     step_counter: int = 0
     current_wave: int = 0