From ab832c4ca7c9fc51fe3ce0d44a05f3f5c853965d Mon Sep 17 00:00:00 2001
From: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
Date: Tue, 25 Aug 2026 15:54:05 -0400
Subject: [PATCH 1/5] [Core][Frontend] Add per-request prefix cache telemetry

Expose request-attributed prompt, prefix-cache, physical block, and prefill chunk metrics through the opt-in per-request metrics surface and existing offline/stat logger outputs.

Assisted-by: OpenAI Codex
Signed-off-by: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
---
 docs/features/per_request_metrics.md          | 50 ++++++++++++-
 .../chat_completion/test_serving_chat.py      | 36 +++++++++-
 .../completion/test_completion_error.py       | 62 +++++++++++++++-
 tests/v1/core/test_prefix_caching.py          | 56 +++++++++++++++
 tests/v1/core/test_scheduler.py               |  9 ++-
 tests/v1/engine/test_output_processor.py      | 53 +++++++++++++-
 tests/v1/metrics/test_stats.py                | 51 ++++++++++++-
 vllm/entrypoints/generate/base/serving.py     | 71 ++++++++++++-------
 .../openai/chat_completion/serving.py         |  8 ++-
 vllm/entrypoints/openai/completion/serving.py | 12 +++-
 vllm/entrypoints/openai/engine/protocol.py    | 17 +++++
 vllm/outputs.py                               | 15 +++-
 vllm/v1/core/block_pool.py                    | 18 +++++
 vllm/v1/core/kv_cache_manager.py              | 30 ++++++++
 vllm/v1/core/sched/scheduler.py               |  5 ++
 vllm/v1/engine/output_processor.py            |  5 ++
 vllm/v1/metrics/stats.py                      | 27 +++++++
 17 files changed, 484 insertions(+), 41 deletions(-)

diff --git a/docs/features/per_request_metrics.md b/docs/features/per_request_metrics.md
index 25211ecfc39b..bf1e3f2447e0 100644
--- a/docs/features/per_request_metrics.md
+++ b/docs/features/per_request_metrics.md
@@ -41,7 +41,19 @@ When per-request metrics are enabled, the response includes a `metrics` object:
     "generation_time_ms": 1240.5,
     "queue_time_ms": 12.3,
     "mean_itl_ms": 9.1,
-    "tokens_per_second": 103.2
+    "tokens_per_second": 103.2,
+    "prefix_cache": {
+      "num_computed_tokens": 26,
+      "num_cached_tokens": 16,
+      "num_local_cached_tokens": 16,
+      "num_external_cached_tokens": 0,
+      "num_cache_creation_tokens": 26,
+      "num_new_full_blocks": 2,
+      "num_block_allocations": 3,
+      "num_block_evictions": 1,
+      "num_prefill_chunks": 1,
+      "prefill_time_ms": 85.2
+    }
   }
 }
 ```
@@ -54,8 +66,40 @@ When per-request metrics are enabled, the response includes a `metrics` object:
 | `mean_itl_ms` | Mean inter-token latency (average time between successive output tokens) during the decode phase. `null` for single-token responses. |
 | `tokens_per_second` | Overall output token throughput: all generated tokens over the inference interval (scheduling to last output token). Unlike `generation_time_ms`, this includes the prefill phase, so it reflects end-to-end generation speed rather than pure decode speed. |
 
-All fields are `null` if the underlying timing data is not available for that
-request.
+All timing fields are `null` if the underlying timestamp data is not available
+for that request.
+
+The experimental `metrics.prefix_cache` object provides request-attributed
+prompt and KV-cache telemetry:
+
+| Field | Description |
+| --- | --- |
+| `num_computed_tokens` | Logical prompt tokens assigned to local model computation at first admission. Recomputation after preemption is not double-counted. |
+| `num_cached_tokens` | Prompt tokens skipped during local computation (`num_local_cached_tokens + num_external_cached_tokens`). |
+| `num_local_cached_tokens` | Prompt tokens supplied by the local prefix cache. |
+| `num_external_cached_tokens` | Prompt tokens supplied through an external KV transfer. This describes the scheduler source, not whether every transferred block was a cache hit in an upstream deployment. |
+| `num_cache_creation_tokens` | Prompt tokens counted as local prefix-cache creation for the request. |
+| `num_new_full_blocks` | Physical KV blocks newly inserted or promoted in local prefix-cache hash maps during prefill. |
+| `num_block_allocations` | Physical KV blocks allocated during prefill. |
+| `num_block_evictions` | Cached physical KV blocks evicted by allocations for this request. This attributes the eviction trigger; it does not claim that the request owned the evicted block. |
+| `num_prefill_chunks` | Scheduler iterations that processed at least one prompt token, including recomputation after preemption. |
+| `prefill_time_ms` | Time from first scheduling to the first output token, including prefill-time preemptions. It currently has the same measurement boundaries as `time_to_first_token_ms`. |
+
+Block counts are physical counts summed across KV cache groups. Consequently,
+one logical token block can contribute more than one physical block on hybrid
+models. The response's top-level `id` correlates the telemetry with the request;
+the same ID is present on the final streaming chunk.
+
+These engine-level fields intentionally live under `metrics`, rather than
+OpenAI `usage`: block allocation and eviction are implementation details, and
+external-transfer and local-cache sources are not billing-token categories.
+
+Offline generation exposes the same engine values as
+`RequestOutput.prefill_stats`, alongside `RequestOutput.request_id`. Custom stat
+logger plugins receive it as `FinishedRequestStats.prefill_stats`, correlated by
+`FinishedRequestStats.request_id`. As documented by the stat-logger interface,
+the plugin-side stats classes are not stable APIs and can change between
+versions.
 
 !!! note
     Timing metrics describe a single generation stream, so they are only
diff --git a/tests/entrypoints/openai/chat_completion/test_serving_chat.py b/tests/entrypoints/openai/chat_completion/test_serving_chat.py
index debf547593fa..b3d69d4748ad 100644
--- a/tests/entrypoints/openai/chat_completion/test_serving_chat.py
+++ b/tests/entrypoints/openai/chat_completion/test_serving_chat.py
@@ -54,7 +54,7 @@
 from vllm.tokenizers.mistral import MistralTokenizer
 from vllm.tokenizers.registry import cached_tokenizer_from_config
 from vllm.v1.engine.async_llm import AsyncLLM
-from vllm.v1.metrics.stats import RequestStateStats
+from vllm.v1.metrics.stats import PrefillStats, RequestStateStats
 
 GPT_OSS_MODEL_NAME = "openai/gpt-oss-20b"
 GPT_OSS_SPECULATOR_NAME = "RedHatAI/gpt-oss-20b-speculator.eagle3"
@@ -65,6 +65,18 @@
     last_token_ts=3.0,
     num_generation_tokens=2,
 )
+_PREFIX_CACHE_STATS = PrefillStats(
+    num_prompt_tokens=42,
+    num_computed_tokens=26,
+    num_cached_tokens=16,
+    num_local_cached_tokens=12,
+    num_external_cached_tokens=4,
+    num_cache_creation_tokens=24,
+    num_new_full_blocks=2,
+    num_block_allocations=3,
+    num_block_evictions=1,
+    num_prefill_chunks=2,
+)
 
 
 @pytest.fixture(scope="module")
@@ -660,6 +672,7 @@ def _make_metrics_request_output(
         ],
         finished=True,
         metrics=metrics,
+        prefill_stats=_PREFIX_CACHE_STATS,
     )
 
 
@@ -710,6 +723,21 @@ def test_build_per_request_timing_metrics_valid_timestamps():
     assert metrics.mean_itl_ms == pytest.approx(1000.0 / 9, rel=1e-4)
     assert metrics.tokens_per_second == pytest.approx(10.0 / 1.5, rel=1e-4)
 
+    metrics = build_per_request_timing_metrics(
+        _PER_REQUEST_STATS,
+        num_generation_tokens=10,
+        prefill_stats=_PREFIX_CACHE_STATS,
+    )
+    assert metrics.prefix_cache is not None
+    assert metrics.prefix_cache.num_computed_tokens == 26
+    assert metrics.prefix_cache.num_local_cached_tokens == 12
+    assert metrics.prefix_cache.num_external_cached_tokens == 4
+    assert metrics.prefix_cache.num_new_full_blocks == 2
+    assert metrics.prefix_cache.num_block_allocations == 3
+    assert metrics.prefix_cache.num_block_evictions == 1
+    assert metrics.prefix_cache.num_prefill_chunks == 2
+    assert metrics.prefix_cache.prefill_time_ms == pytest.approx(500.0)
+
 
 @pytest.mark.asyncio
 async def test_chat_per_request_metrics_follow_server_flag():
@@ -750,6 +778,9 @@ async def test_chat_per_request_metrics_follow_server_flag():
     )
     assert enabled_response.metrics is not None
     assert enabled_response.metrics.time_to_first_token_ms == pytest.approx(500.0)
+    assert enabled_response.id == "chatcmpl-test-id"
+    assert enabled_response.metrics.prefix_cache is not None
+    assert enabled_response.metrics.prefix_cache.num_cached_tokens == 16
 
 
 @pytest.mark.asyncio
@@ -790,6 +821,9 @@ async def test_chat_streaming_metrics_ride_on_usage_chunk():
     usage_chunks = [chunk for chunk in chunks if chunk.get("usage")]
     assert usage_chunks
     assert usage_chunks[-1]["metrics"]["time_to_first_token_ms"] == pytest.approx(500.0)
+    assert usage_chunks[-1]["id"] == "chatcmpl-test-id"
+    assert usage_chunks[-1]["metrics"]["prefix_cache"]["num_computed_tokens"] == 26
+    assert usage_chunks[-1]["metrics"]["prefix_cache"]["num_block_evictions"] == 1
 
 
 @pytest.mark.asyncio
diff --git a/tests/entrypoints/openai/completion/test_completion_error.py b/tests/entrypoints/openai/completion/test_completion_error.py
index 56a2ba66f1ad..7fad319a63e0 100644
--- a/tests/entrypoints/openai/completion/test_completion_error.py
+++ b/tests/entrypoints/openai/completion/test_completion_error.py
@@ -1,6 +1,7 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
 
+import json
 from dataclasses import dataclass, field
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
@@ -23,7 +24,11 @@
 from vllm.renderers.online_renderer import OnlineRenderer
 from vllm.tokenizers.registry import cached_tokenizer_from_config
 from vllm.v1.engine.async_llm import AsyncLLM
-from vllm.v1.metrics.stats import RequestSpecDecodeMetrics, RequestStateStats
+from vllm.v1.metrics.stats import (
+    PrefillStats,
+    RequestSpecDecodeMetrics,
+    RequestStateStats,
+)
 
 MODEL_NAME = "openai-community/gpt2"
 MODEL_NAME_SHORT = "gpt2"
@@ -34,6 +39,18 @@
     last_token_ts=3.0,
     num_generation_tokens=2,
 )
+_PREFIX_CACHE_STATS = PrefillStats(
+    num_prompt_tokens=42,
+    num_computed_tokens=26,
+    num_cached_tokens=16,
+    num_local_cached_tokens=12,
+    num_external_cached_tokens=4,
+    num_cache_creation_tokens=24,
+    num_new_full_blocks=2,
+    num_block_allocations=3,
+    num_block_evictions=1,
+    num_prefill_chunks=2,
+)
 BASE_MODEL_PATHS = [
     BaseModelPath(name=MODEL_NAME, model_path=MODEL_NAME),
     BaseModelPath(name=MODEL_NAME_SHORT, model_path=MODEL_NAME_SHORT),
@@ -109,6 +126,8 @@ def _build_minimal_metrics_serving_completion(
 ) -> OpenAIServingCompletion:
     serving = OpenAIServingCompletion.__new__(OpenAIServingCompletion)
     serving.enable_prompt_tokens_details = False
+    serving.enable_force_include_usage = False
+    serving.return_tokens_as_token_ids = False
     serving.system_fingerprint = None
     serving.enable_per_request_metrics = enable_per_request_metrics
     return serving
@@ -134,6 +153,7 @@ def _make_metrics_request_output(
         ],
         finished=True,
         metrics=metrics,
+        prefill_stats=_PREFIX_CACHE_STATS,
     )
 
 
@@ -176,6 +196,46 @@ def test_completion_per_request_metrics_follow_server_flag():
     )
     assert enabled_response.metrics is not None
     assert enabled_response.metrics.time_to_first_token_ms == pytest.approx(500.0)
+    assert enabled_response.id == "cmpl-test-id"
+    assert enabled_response.metrics.prefix_cache is not None
+    assert enabled_response.metrics.prefix_cache.num_external_cached_tokens == 4
+
+
+@pytest.mark.asyncio
+async def test_completion_streaming_prefix_cache_metrics_ride_on_usage_chunk():
+    serving = _build_minimal_metrics_serving_completion(enable_per_request_metrics=True)
+    request = CompletionRequest(
+        model=MODEL_NAME,
+        prompt="Test prompt",
+        max_tokens=10,
+        stream=True,
+        stream_options={"include_usage": True},
+    )
+
+    async def result_generator():
+        yield 0, _make_metrics_request_output()
+
+    chunks = []
+    async for line in serving.completion_stream_generator(
+        request=request,
+        engine_inputs=[MagicMock()],
+        result_generator=result_generator(),
+        request_id="cmpl-test-id",
+        created_time=0,
+        model_name=MODEL_NAME,
+        num_prompts=1,
+        tokenizer=None,
+        request_metadata=RequestResponseMetadata(request_id="cmpl-test-id"),
+    ):
+        payload = line.removeprefix("data: ").strip()
+        if payload != "[DONE]":
+            chunks.append(json.loads(payload))
+
+    usage_chunks = [chunk for chunk in chunks if chunk.get("usage")]
+    assert usage_chunks[-1]["id"] == "cmpl-test-id"
+    prefix_cache = usage_chunks[-1]["metrics"]["prefix_cache"]
+    assert prefix_cache["num_cached_tokens"] == 16
+    assert prefix_cache["num_prefill_chunks"] == 2
 
 
 def test_completion_per_request_metrics_suppressed_for_multiple_prompts():
diff --git a/tests/v1/core/test_prefix_caching.py b/tests/v1/core/test_prefix_caching.py
index 6fc2131b6e87..77d36b396283 100644
--- a/tests/v1/core/test_prefix_caching.py
+++ b/tests/v1/core/test_prefix_caching.py
@@ -2010,6 +2010,62 @@ def test_prefix_cache_stats_disabled():
     assert manager.prefix_cache_stats is None
 
 
+def test_prefill_block_activity_is_attributed_to_allocating_request():
+    block_size = 4
+    manager = make_kv_cache_manager(
+        make_kv_cache_config(block_size, 4),
+        max_model_len=8192,
+        enable_caching=True,
+        hash_block_size=block_size,
+        log_stats=True,
+    )
+
+    first = make_request("first", list(range(8)), block_size, sha256)
+    assert manager.allocate_slots(first, 8) is not None
+    assert first.prefill_stats is not None
+    assert first.prefill_stats.num_block_allocations == 2
+    assert first.prefill_stats.num_block_evictions == 0
+    assert first.prefill_stats.num_new_full_blocks == 2
+    manager.free(first)
+
+    # Three physical blocks remain after the null block. Allocating all three
+    # for a different request reuses one uncached block and evicts the two
+    # cached blocks created by ``first``.
+    second = make_request("second", list(range(100, 112)), block_size, sha256)
+    assert manager.allocate_slots(second, 12) is not None
+    assert second.prefill_stats is not None
+    assert second.prefill_stats.num_block_allocations == 3
+    assert second.prefill_stats.num_block_evictions == 2
+    assert second.prefill_stats.num_new_full_blocks == 3
+
+    # Attribution is immutable after the first request's prefill operations.
+    assert first.prefill_stats.num_block_allocations == 2
+    assert first.prefill_stats.num_block_evictions == 0
+
+
+def test_delayed_cache_creation_is_attributed_to_request_prefill():
+    block_size = 4
+    manager = make_kv_cache_manager(
+        make_kv_cache_config(block_size, 4),
+        max_model_len=8192,
+        enable_caching=True,
+        hash_block_size=block_size,
+        log_stats=True,
+    )
+    request = make_request("async-load", list(range(8)), block_size, sha256)
+
+    assert manager.allocate_slots(request, 8, delay_cache_blocks=True) is not None
+    assert request.prefill_stats is not None
+    assert request.prefill_stats.num_block_allocations == 2
+    assert request.prefill_stats.num_new_full_blocks == 0
+
+    # Async KV paths commit block hashes after the transfer completes, outside
+    # allocate_slots(). The later cache operation must retain request attribution.
+    manager.cache_blocks(request, 8)
+    assert request.prefill_stats.num_new_full_blocks == 2
+    assert request.prefill_stats.num_block_allocations == 2
+
+
 def test_maybe_evict_cached_block():
     pool = BlockPool(num_gpu_blocks=4, enable_caching=True, hash_block_size=16)
     block_hash0 = make_block_hash_with_group_id(BlockHash(b"10"), 1000)
diff --git a/tests/v1/core/test_scheduler.py b/tests/v1/core/test_scheduler.py
index 920823baeb8d..b77aed676a6c 100644
--- a/tests/v1/core/test_scheduler.py
+++ b/tests/v1/core/test_scheduler.py
@@ -1613,6 +1613,8 @@ def test_no_spec_tokens_scheduled_for_prefill_chunks():
     output = scheduler.schedule()
     assert len(output.scheduled_new_reqs) == 1
     assert output.num_scheduled_tokens[req.request_id] == 50
+    assert req.prefill_stats is not None
+    assert req.prefill_stats.num_prefill_chunks == 1
 
     # Update from output (no sampled token since still prefilling)
     req_to_index = {req.request_id: 0}
@@ -1644,6 +1646,8 @@ def test_no_spec_tokens_scheduled_for_prefill_chunks():
     assert req.request_id not in output.scheduled_spec_decode_tokens, (
         "Spec tokens should not be scheduled with prefill chunks"
     )
+    assert req.prefill_stats is not None
+    assert req.prefill_stats.num_prefill_chunks == 2
 
     # Update from output with a sampled token (prefill complete)
     model_runner_output = ModelRunnerOutput(
@@ -1654,7 +1658,10 @@ def test_no_spec_tokens_scheduled_for_prefill_chunks():
         prompt_logprobs_dict={},
         pooler_output=[],
     )
-    scheduler.update_from_output(output, model_runner_output)
+    engine_core_outputs = scheduler.update_from_output(output, model_runner_output)
+    prefill_stats = engine_core_outputs[0].outputs[0].prefill_stats
+    assert prefill_stats is not None
+    assert prefill_stats.num_prefill_chunks == 2
 
     # Now provide draft tokens - should be accepted since prefill is complete
     draft_token_ids = DraftTokenIds([req.request_id], [[1, 2, 3]])
diff --git a/tests/v1/engine/test_output_processor.py b/tests/v1/engine/test_output_processor.py
index fa63b52200fc..c3161ea0c05e 100644
--- a/tests/v1/engine/test_output_processor.py
+++ b/tests/v1/engine/test_output_processor.py
@@ -22,12 +22,13 @@
 from vllm.v1.engine import (
     EngineCoreEvent,
     EngineCoreEventType,
+    EngineCoreOutput,
     EngineCoreOutputs,
     EngineCoreRequest,
     FinishReason,
 )
 from vllm.v1.engine.output_processor import OutputProcessor, RequestOutputCollector
-from vllm.v1.metrics.stats import IterationStats, SchedulerStats
+from vllm.v1.metrics.stats import IterationStats, PrefillStats, SchedulerStats
 
 
 def _ref_convert_id_to_token(
@@ -46,6 +47,56 @@ def _ref_convert_id_to_token(
     return tokenizer.decode([token_id]) or ""
 
 
+def test_prefill_stats_propagate_to_request_output():
+    output_processor = OutputProcessor(tokenizer=None, log_stats=False)
+    request = EngineCoreRequest(
+        request_id="request-internal",
+        external_req_id="request-external",
+        prompt_token_ids=[1, 2, 3, 4],
+        mm_features=None,
+        arrival_time=0,
+        lora_request=None,
+        cache_salt=None,
+        data_parallel_rank=None,
+        sampling_params=SamplingParams(
+            detokenize=False,
+            output_kind=RequestOutputKind.DELTA,
+        ),
+        pooling_params=None,
+    )
+    output_processor.add_request(request, prompt=None)
+
+    prefill_stats = PrefillStats(
+        num_prompt_tokens=4,
+        num_computed_tokens=2,
+        num_cached_tokens=2,
+        num_local_cached_tokens=1,
+        num_external_cached_tokens=1,
+        num_cache_creation_tokens=2,
+        num_new_full_blocks=1,
+        num_block_allocations=2,
+        num_block_evictions=1,
+        num_prefill_chunks=2,
+    )
+    processed = output_processor.process_outputs(
+        [
+            EngineCoreOutput(
+                request_id=request.request_id,
+                new_token_ids=[42],
+                finish_reason=FinishReason.LENGTH,
+                prefill_stats=prefill_stats,
+            )
+        ]
+    )
+
+    assert len(processed.request_outputs) == 1
+    request_output = processed.request_outputs[0]
+    assert isinstance(request_output, RequestOutput)
+    assert request_output.request_id == "request-external"
+    assert request_output.prefill_stats is prefill_stats
+    assert request_output.prefill_stats.num_block_evictions == 1
+
+
 @pytest.mark.parametrize(
     "request_output_kind", [RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY]
 )
diff --git a/tests/v1/metrics/test_stats.py b/tests/v1/metrics/test_stats.py
index 0de74f0faaa4..44a83475e6c3 100644
--- a/tests/v1/metrics/test_stats.py
+++ b/tests/v1/metrics/test_stats.py
@@ -1,7 +1,9 @@
 # SPDX-License-Identifier: Apache-2.0
 # SPDX-FileCopyrightText: Copyright contributors to the vLLM project
+import pytest
+
 from vllm.v1.core.sched.output import ScheduledEncoderInputStats, SchedulerOutput
-from vllm.v1.engine import EngineCoreOutputs, FinishReason
+from vllm.v1.engine import EngineCoreOutput, EngineCoreOutputs, FinishReason
 from vllm.v1.metrics.stats import (
     IterationStats,
     PrefillStats,
@@ -45,6 +47,36 @@ def test_scheduler_iteration_details_serialization():
     assert decoded.scheduler_stats.iteration_details == iteration_details
 
 
+def test_prefill_stats_serialization_includes_request_kv_activity():
+    prefill_stats = PrefillStats(
+        num_prompt_tokens=64,
+        num_computed_tokens=32,
+        num_cached_tokens=32,
+        num_local_cached_tokens=24,
+        num_external_cached_tokens=8,
+        num_cache_creation_tokens=16,
+        num_new_full_blocks=2,
+        num_block_allocations=3,
+        num_block_evictions=1,
+        num_prefill_chunks=2,
+    )
+    outputs = EngineCoreOutputs(
+        outputs=[
+            EngineCoreOutput(
+                request_id="request-123",
+                new_token_ids=[42],
+                prefill_stats=prefill_stats,
+            )
+        ]
+    )
+
+    encoded = MsgpackEncoder().encode(outputs)
+    decoded = MsgpackDecoder(EngineCoreOutputs).decode(encoded)
+
+    assert decoded.outputs[0].request_id == "request-123"
+    assert decoded.outputs[0].prefill_stats == prefill_stats
+
+
 def test_compute_iteration_details_includes_encoder_stats():
     scheduler_output = SchedulerOutput.make_empty()
     scheduler_output.scheduled_encoder_input_stats = ScheduledEncoderInputStats(
@@ -68,6 +100,18 @@ def test_prefill_kv_computed_with_cache():
     req_stats.num_generation_tokens = 50
 
     # Case 1: With prefix cache (1200 tokens cached)
+    prefill_stats = PrefillStats(
+        num_prompt_tokens=10000,
+        num_computed_tokens=8800,
+        num_cached_tokens=1200,
+        num_local_cached_tokens=1000,
+        num_external_cached_tokens=200,
+        num_cache_creation_tokens=8000,
+        num_new_full_blocks=550,
+        num_block_allocations=551,
+        num_block_evictions=3,
+        num_prefill_chunks=4,
+    )
     iteration_stats.update_from_finished_request(
         finish_reason=FinishReason.STOP,
         request_id="test-req-001",
@@ -75,12 +119,17 @@ def test_prefill_kv_computed_with_cache():
         max_tokens_param=100,
         req_stats=req_stats,
         num_cached_tokens=1200,
+        prefill_stats=prefill_stats,
     )
 
     finished_req = iteration_stats.finished_requests[0]
     assert finished_req.num_prompt_tokens == 10000
     assert finished_req.num_cached_tokens == 1200
     assert finished_req.request_id == "test-req-001"
+    assert finished_req.prefill_stats is prefill_stats
+    assert finished_req.prefill_stats.num_external_cached_tokens == 200
+    assert finished_req.prefill_stats.num_block_evictions == 3
+    assert finished_req.prefill_time == pytest.approx(0.4)
 
     # Verify calculation: prefill KV = prompt tokens - cached tokens
     prefill_kv_computed = finished_req.num_prompt_tokens - max(
diff --git a/vllm/entrypoints/generate/base/serving.py b/vllm/entrypoints/generate/base/serving.py
index d6ae0a209065..861f87c5757b 100644
--- a/vllm/entrypoints/generate/base/serving.py
+++ b/vllm/entrypoints/generate/base/serving.py
@@ -19,6 +19,7 @@
     ErrorResponse,
     GenerationError,
     PerRequestMetrics,
+    PrefixCacheMetrics,
     SpeculativeDecodingMetrics,
 )
 from vllm.entrypoints.openai.models.serving import OpenAIServingModels
@@ -36,7 +37,7 @@
     extract_trace_headers,
     log_tracing_disabled_warning,
 )
-from vllm.v1.metrics.stats import RequestStateStats
+from vllm.v1.metrics.stats import PrefillStats, RequestStateStats
 
 if TYPE_CHECKING:
     from vllm.outputs import RequestOutput
@@ -52,6 +53,7 @@
 def build_per_request_timing_metrics(
     metrics: RequestStateStats | None,
     num_generation_tokens: int,
+    prefill_stats: PrefillStats | None = None,
 ) -> PerRequestMetrics:
     """Build per-request timing metrics from ``RequestStateStats``.
 
@@ -61,39 +63,53 @@ def build_per_request_timing_metrics(
     over the inference interval (scheduling to last output token), so it counts
     the prefill/TTFT phase and is not simply the reciprocal of ``mean_itl_ms``.
     Each field is left ``None`` when the timestamps it depends on are
-    unavailable.
+    unavailable. When ``prefill_stats`` is available, the experimental
+    prompt/KV breakdown is attached as ``prefix_cache``.
     """
-    if metrics is None:
-        return PerRequestMetrics()
-
-    queued_ts = metrics.queued_ts
-    scheduled_ts = metrics.scheduled_ts
-    first_token_ts = metrics.first_token_ts
-    last_token_ts = metrics.last_token_ts
-
     time_to_first_token_ms: float | None = None
     generation_time_ms: float | None = None
     queue_time_ms: float | None = None
     mean_itl_ms: float | None = None
     tokens_per_second: float | None = None
 
-    if scheduled_ts > 0 and first_token_ts > 0:
-        time_to_first_token_ms = (first_token_ts - scheduled_ts) * 1000
-
-    if first_token_ts > 0 and last_token_ts > 0:
-        generation_time_ms = (last_token_ts - first_token_ts) * 1000
-
-    if queued_ts > 0 and scheduled_ts > 0:
-        queue_time_ms = (scheduled_ts - queued_ts) * 1000
-
-    if first_token_ts > 0 and last_token_ts > 0 and num_generation_tokens > 1:
-        decode_time = last_token_ts - first_token_ts
-        mean_itl_ms = decode_time / (num_generation_tokens - 1) * 1000
-
-    if scheduled_ts > 0 and last_token_ts > 0:
-        inference_time_ms = (last_token_ts - scheduled_ts) * 1000
-        if inference_time_ms > 0:
-            tokens_per_second = num_generation_tokens / inference_time_ms * 1000
+    if metrics is not None:
+        queued_ts = metrics.queued_ts
+        scheduled_ts = metrics.scheduled_ts
+        first_token_ts = metrics.first_token_ts
+        last_token_ts = metrics.last_token_ts
+
+        if scheduled_ts > 0 and first_token_ts > 0:
+            time_to_first_token_ms = (first_token_ts - scheduled_ts) * 1000
+
+        if first_token_ts > 0 and last_token_ts > 0:
+            generation_time_ms = (last_token_ts - first_token_ts) * 1000
+
+        if queued_ts > 0 and scheduled_ts > 0:
+            queue_time_ms = (scheduled_ts - queued_ts) * 1000
+
+        if first_token_ts > 0 and last_token_ts > 0 and num_generation_tokens > 1:
+            decode_time = last_token_ts - first_token_ts
+            mean_itl_ms = decode_time / (num_generation_tokens - 1) * 1000
+
+        if scheduled_ts > 0 and last_token_ts > 0:
+            inference_time_ms = (last_token_ts - scheduled_ts) * 1000
+            if inference_time_ms > 0:
+                tokens_per_second = num_generation_tokens / inference_time_ms * 1000
+
+    prefix_cache_metrics = None
+    if prefill_stats is not None:
+        prefix_cache_metrics = PrefixCacheMetrics(
+            num_computed_tokens=prefill_stats.num_computed_tokens,
+            num_cached_tokens=prefill_stats.num_cached_tokens,
+            num_local_cached_tokens=prefill_stats.num_local_cached_tokens,
+            num_external_cached_tokens=prefill_stats.num_external_cached_tokens,
+            num_cache_creation_tokens=prefill_stats.num_cache_creation_tokens,
+            num_new_full_blocks=prefill_stats.num_new_full_blocks,
+            num_block_allocations=prefill_stats.num_block_allocations,
+            num_block_evictions=prefill_stats.num_block_evictions,
+            num_prefill_chunks=prefill_stats.num_prefill_chunks,
+            prefill_time_ms=time_to_first_token_ms,
+        )
 
     return PerRequestMetrics(
         time_to_first_token_ms=time_to_first_token_ms,
@@ -101,6 +117,7 @@ def build_per_request_timing_metrics(
         queue_time_ms=queue_time_ms,
         mean_itl_ms=mean_itl_ms,
         tokens_per_second=tokens_per_second,
+        prefix_cache=prefix_cache_metrics,
     )
 
 
diff --git a/vllm/entrypoints/openai/chat_completion/serving.py b/vllm/entrypoints/openai/chat_completion/serving.py
index 8259ef65f4c8..23a66cad2382 100644
--- a/vllm/entrypoints/openai/chat_completion/serving.py
+++ b/vllm/entrypoints/openai/chat_completion/serving.py
@@ -848,7 +848,9 @@ async def chat_completion_stream_generator(
                             last_res.metrics if last_res is not None else None
                         )
                         stream_per_request_metrics = build_per_request_timing_metrics(
-                            last_metrics, completion_tokens
+                            last_metrics,
+                            completion_tokens,
+                            last_res.prefill_stats if last_res is not None else None,
                         )
                     spec_stats = build_spec_decoding_metrics(last_res)
                     if spec_stats is not None:
@@ -1151,7 +1153,9 @@ async def chat_completion_full_generator(
         if (request.n or 1) == 1:
             if self.enable_per_request_metrics:
                 per_request_metrics = build_per_request_timing_metrics(
-                    final_res.metrics, num_generated_tokens
+                    final_res.metrics,
+                    num_generated_tokens,
+                    final_res.prefill_stats,
                 )
             spec_stats = build_spec_decoding_metrics(final_res)
             if spec_stats is not None:
diff --git a/vllm/entrypoints/openai/completion/serving.py b/vllm/entrypoints/openai/completion/serving.py
index 93a2b7d3aa6a..9e0ff4cae624 100644
--- a/vllm/entrypoints/openai/completion/serving.py
+++ b/vllm/entrypoints/openai/completion/serving.py
@@ -470,7 +470,9 @@ async def completion_stream_generator(
                             last_res.metrics if last_res is not None else None
                         )
                         stream_per_request_metrics = build_per_request_timing_metrics(
-                            last_metrics, total_completion_tokens
+                            last_metrics,
+                            total_completion_tokens,
+                            last_res.prefill_stats if last_res is not None else None,
                         )
                     spec_stats = build_spec_decoding_metrics(last_res)
                     if spec_stats is not None:
@@ -628,7 +630,13 @@ def request_output_to_completion_response(
                     last_final_res.metrics if last_final_res is not None else None
                 )
                 per_request_metrics = build_per_request_timing_metrics(
-                    last_metrics, num_generated_tokens
+                    last_metrics,
+                    num_generated_tokens,
+                    (
+                        last_final_res.prefill_stats
+                        if last_final_res is not None
+                        else None
+                    ),
                 )
             spec_stats = build_spec_decoding_metrics(last_final_res)
             if spec_stats is not None:
diff --git a/vllm/entrypoints/openai/engine/protocol.py b/vllm/entrypoints/openai/engine/protocol.py
index 9635ece46b03..58a0a9b1939e 100644
--- a/vllm/entrypoints/openai/engine/protocol.py
+++ b/vllm/entrypoints/openai/engine/protocol.py
@@ -151,6 +151,21 @@ class SpeculativeDecodingMetrics(OpenAIBaseModel):
     per_step_drafted: list[int] | None = None
 
 
+class PrefixCacheMetrics(OpenAIBaseModel):
+    """Experimental per-request prompt/KV telemetry."""
+
+    num_computed_tokens: int
+    num_cached_tokens: int
+    num_local_cached_tokens: int
+    num_external_cached_tokens: int
+    num_cache_creation_tokens: int
+    num_new_full_blocks: int
+    num_block_allocations: int
+    num_block_evictions: int
+    num_prefill_chunks: int
+    prefill_time_ms: float | None = None
+
+
 class PerRequestMetrics(OpenAIBaseModel):
     time_to_first_token_ms: float | None = None
     generation_time_ms: float | None = None
@@ -158,6 +173,8 @@ class PerRequestMetrics(OpenAIBaseModel):
     mean_itl_ms: float | None = None
     tokens_per_second: float | None = None
     # Experimental, subject to change.
+    prefix_cache: PrefixCacheMetrics | None = None
+    # Experimental, subject to change.
     speculative_decoding: SpeculativeDecodingMetrics | None = None
 
 
diff --git a/vllm/outputs.py b/vllm/outputs.py
index 84b9fccdc88d..dbce4c25648f 100644
--- a/vllm/outputs.py
+++ b/vllm/outputs.py
@@ -13,7 +13,11 @@
 from vllm.logger import init_logger
 from vllm.logprobs import PromptLogprobs, SampleLogprobs
 from vllm.lora.request import LoRARequest
-from vllm.v1.metrics.stats import RequestSpecDecodeMetrics, RequestStateStats
+from vllm.v1.metrics.stats import (
+    PrefillStats,
+    RequestSpecDecodeMetrics,
+    RequestStateStats,
+)
 
 logger = init_logger(__name__)
 
@@ -126,6 +130,8 @@ class RequestOutput:
         num_cached_tokens: The number of tokens with prefix cache hit.
         num_cache_creation_tokens: Prompt tokens currently counted as local
             prefix-cache writes for this request.
+        prefill_stats: Per-request prompt/KV telemetry emitted when prefill
+            completes.
         kv_transfer_params: The params for remote K/V transfer.
         ec_transfer_params: The params for remote encoder-cache transfer.
     """
@@ -145,6 +151,7 @@ def __init__(
         num_cached_tokens: int | None = None,
         num_cache_creation_tokens: int | None = None,
         *,
+        prefill_stats: PrefillStats | None = None,
         kv_transfer_params: dict[str, Any] | None = None,
         ec_transfer_params: dict[str, Any] | None = None,
         # Forward compatibility, code that uses args added in new release can
@@ -167,6 +174,7 @@ def __init__(
         self.encoder_prompt_token_ids = encoder_prompt_token_ids
         self.num_cached_tokens = num_cached_tokens
         self.num_cache_creation_tokens = num_cache_creation_tokens
+        self.prefill_stats = prefill_stats
         self.kv_transfer_params = kv_transfer_params
         self.ec_transfer_params = ec_transfer_params
 
@@ -176,6 +184,8 @@ def add(self, next_output: "RequestOutput", aggregate: bool) -> None:
         self.finished |= next_output.finished
         self.kv_transfer_params = next_output.kv_transfer_params
         self.ec_transfer_params = next_output.ec_transfer_params
+        if self.prefill_stats is None:
+            self.prefill_stats = next_output.prefill_stats
 
         for next_completion in next_output.outputs:
             for i, completion in enumerate(self.outputs):
@@ -214,7 +224,8 @@ def __repr__(self) -> str:
             f"metrics={self.metrics}, "
             f"lora_request={self.lora_request}, "
             f"num_cached_tokens={self.num_cached_tokens}, "
-            f"num_cache_creation_tokens={self.num_cache_creation_tokens})"
+            f"num_cache_creation_tokens={self.num_cache_creation_tokens}, "
+            f"prefill_stats={self.prefill_stats})"
         )
 
 
diff --git a/vllm/v1/core/block_pool.py b/vllm/v1/core/block_pool.py
index a8a6c1ca94da..f126781fc589 100644
--- a/vllm/v1/core/block_pool.py
+++ b/vllm/v1/core/block_pool.py
@@ -195,6 +195,21 @@ def __init__(
 
         self.metrics_collector = metrics_collector
 
+        # Monotonic counters used to attribute exact physical block activity to
+        # a request by taking deltas around the scheduler's synchronous cache
+        # operations. These counters do not participate in cache decisions.
+        self._num_block_allocations = 0
+        self._num_block_evictions = 0
+        self._num_new_full_blocks = 0
+
+    def get_block_activity(self) -> tuple[int, int, int]:
+        """Return cumulative (allocations, evictions, new full blocks)."""
+        return (
+            self._num_block_allocations,
+            self._num_block_evictions,
+            self._num_new_full_blocks,
+        )
+
     def get_cached_block(
         self, block_hash: BlockHash, kv_cache_group_ids: list[int]
     ) -> list[KVCacheBlock] | None:
@@ -295,6 +310,7 @@ def cache_full_blocks(
                 blk,
                 num_tokens=num_hash_tokens,
             )
+            self._num_new_full_blocks += 1
             if new_hashes is not None:
                 new_hashes.append(maybe_convert_block_hash(block_hash))
 
@@ -659,6 +675,7 @@ def get_new_blocks(self, num_blocks: int) -> list[KVCacheBlock]:
             raise ValueError(f"Cannot get {num_blocks} free blocks from the pool")
 
         ret: list[KVCacheBlock] = self.free_block_queue.popleft_n(num_blocks)
+        self._num_block_allocations += len(ret)
 
         # In order to only iterate the list once, we duplicated code a bit
         if self.enable_caching:
@@ -697,6 +714,7 @@ def _maybe_evict_cached_block(self, block: KVCacheBlock) -> bool:
             return False
 
         self._emit_block_removed_events(evicted_hashes)
+        self._num_block_evictions += 1
         return True
 
     def touch(self, blocks: Sequence[KVCacheBlock]) -> None:
diff --git a/vllm/v1/core/kv_cache_manager.py b/vllm/v1/core/kv_cache_manager.py
index d1af91b65993..03563cc90d64 100644
--- a/vllm/v1/core/kv_cache_manager.py
+++ b/vllm/v1/core/kv_cache_manager.py
@@ -448,6 +448,12 @@ def allocate_slots(
                 "external computed tokens"
             )
 
+        block_activity_before = (
+            self.block_pool.get_block_activity()
+            if request.prefill_stats is not None
+            else None
+        )
+
         if new_computed_blocks is not None:
             new_computed_block_list = new_computed_blocks.blocks
         else:
@@ -552,6 +558,7 @@ def allocate_slots(
         # P/D: delay caching blocks if we have to recv from
         # remote. Update state for locally cached blocks.
         if not self.enable_caching or delay_cache_blocks:
+            self._record_prefill_block_activity(request, block_activity_before)
             return self.create_kv_cache_blocks(new_blocks)
 
         # NOTE(woosuk): We want to commit (cache) up to num_local_computed_tokens
@@ -565,8 +572,25 @@ def allocate_slots(
         )
         self.coordinator.cache_blocks(request, num_tokens_to_cache)
 
+        self._record_prefill_block_activity(request, block_activity_before)
+
         return self.create_kv_cache_blocks(new_blocks)
 
+    def _record_prefill_block_activity(
+        self,
+        request: Request,
+        before: tuple[int, int, int] | None,
+    ) -> None:
+        """Record block-pool deltas while a request's prefill is in flight."""
+        if before is None or request.prefill_stats is None:
+            return
+        after = self.block_pool.get_block_activity()
+        request.prefill_stats.record_block_activity(
+            num_block_allocations=after[0] - before[0],
+            num_block_evictions=after[1] - before[1],
+            num_new_full_blocks=after[2] - before[2],
+        )
+
     def free(self, request: Request) -> None:
         """Free the blocks allocated for the request.
         We free the blocks in reverse order so that the tail blocks are evicted
@@ -769,7 +793,13 @@ def cache_blocks(self, request: Request, num_computed_tokens: int) -> None:
                 that are already cached and tokens to be cached.
         """
         if self.enable_caching:
+            block_activity_before = (
+                self.block_pool.get_block_activity()
+                if request.prefill_stats is not None
+                else None
+            )
             self.coordinator.cache_blocks(request, num_computed_tokens)
+            self._record_prefill_block_activity(request, block_activity_before)
 
     def create_kv_cache_blocks(
         self, blocks: tuple[list[KVCacheBlock], ...]
diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py
index c373d5cf5c87..c647bb941ec6 100644
--- a/vllm/v1/core/sched/scheduler.py
+++ b/vllm/v1/core/sched/scheduler.py
@@ -1418,6 +1418,11 @@ def _update_after_schedule(self, scheduler_output: SchedulerOutput) -> None:
         num_scheduled_tokens = scheduler_output.num_scheduled_tokens
         for req_id, num_scheduled_token in num_scheduled_tokens.items():
             request = self.requests[req_id]
+            if (
+                request.prefill_stats is not None
+                and request.num_computed_tokens < request.num_prompt_tokens
+            ):
+                request.prefill_stats.num_prefill_chunks += 1
             request.num_computed_tokens += num_scheduled_token
             request.num_in_flight_tokens += num_scheduled_token
             if self.defer_block_free:
diff --git a/vllm/v1/engine/output_processor.py b/vllm/v1/engine/output_processor.py
index 055e3af574c1..ef2d1011d9c0 100644
--- a/vllm/v1/engine/output_processor.py
+++ b/vllm/v1/engine/output_processor.py
@@ -35,6 +35,7 @@
 from vllm.v1.metrics.stats import (
     IterationStats,
     LoRARequestStates,
+    PrefillStats,
     RequestSpecDecodeMetrics,
     RequestStateStats,
     SchedulerStats,
@@ -176,6 +177,7 @@ def __init__(
         self.queue = queue
         self.num_cached_tokens = 0
         self.num_cache_creation_tokens = 0
+        self.prefill_stats: PrefillStats | None = None
         # Per-sequence spec-decode accumulator; arrives once (on finish) via
         # EngineCoreOutput, then attached to this sequence's CompletionOutput.
         self.spec_decode_metrics: RequestSpecDecodeMetrics | None = None
@@ -389,6 +391,7 @@ def _new_request_output(
             ec_transfer_params=ec_transfer_params,
             num_cached_tokens=self.num_cached_tokens,
             num_cache_creation_tokens=self.num_cache_creation_tokens,
+            prefill_stats=self.prefill_stats,
             metrics=self.stats,
         )
 
@@ -655,6 +658,7 @@ def process_outputs(
 
             if req_state.is_prefilling:
                 if engine_core_output.prefill_stats is not None:
+                    req_state.prefill_stats = engine_core_output.prefill_stats
                     req_state.num_cached_tokens = (
                         engine_core_output.prefill_stats.num_cached_tokens
                     )
@@ -849,6 +853,7 @@ def _update_stats_from_finished(
             max_tokens_param=req_state.max_tokens_param,
             req_stats=req_state.stats,
             num_cached_tokens=req_state.num_cached_tokens,
+            prefill_stats=req_state.prefill_stats,
         )
         self.lora_states.request_finished(req_state.request_id, req_state.lora_name)
 
diff --git a/vllm/v1/metrics/stats.py b/vllm/v1/metrics/stats.py
index 3dbc5206ca94..6e36c41624ab 100644
--- a/vllm/v1/metrics/stats.py
+++ b/vllm/v1/metrics/stats.py
@@ -253,6 +253,7 @@ class FinishedRequestStats:
     mean_time_per_output_token: float = 0.0
     is_corrupted: bool = False
     num_cached_tokens: int = 0
+    prefill_stats: "PrefillStats | None" = None
 
 
 @dataclass
@@ -266,6 +267,15 @@ class PrefillStats:
         num_local_cached_tokens: Tokens to be prefilled from local prefix cache.
         num_external_cached_tokens: Tokens to be prefilled from external KV transfer.
         num_cache_creation_tokens: Tokens computed and written to the prefix cache.
+        num_new_full_blocks: Physical KV blocks newly inserted or promoted in the
+            local prefix cache, summed across KV cache groups.
+        num_block_allocations: Physical KV blocks allocated while prefilling,
+            summed across KV cache groups.
+        num_block_evictions: Cached physical KV blocks evicted by those
+            allocations. This attributes the eviction trigger, not ownership of
+            the evicted block, to the request.
+        num_prefill_chunks: Scheduler iterations that process at least one prompt
+            token for the request, including recomputation after preemption.
     """
 
     num_prompt_tokens: int = 0
@@ -274,6 +284,10 @@ class PrefillStats:
     num_local_cached_tokens: int = 0
     num_external_cached_tokens: int = 0
     num_cache_creation_tokens: int = 0
+    num_new_full_blocks: int = 0
+    num_block_allocations: int = 0
+    num_block_evictions: int = 0
+    num_prefill_chunks: int = 0
 
     def set(
         self,
@@ -296,6 +310,17 @@ def finalize(self, num_cached_tokens: int) -> None:
             0, min(num_cached_tokens, self.num_prompt_tokens) - self.num_cached_tokens
         )
 
+    def record_block_activity(
+        self,
+        num_block_allocations: int,
+        num_block_evictions: int,
+        num_new_full_blocks: int,
+    ) -> None:
+        """Accumulate physical block activity attributed to this prefill."""
+        self.num_block_allocations += num_block_allocations
+        self.num_block_evictions += num_block_evictions
+        self.num_new_full_blocks += num_new_full_blocks
+
 
 @dataclass
 class RequestSpecDecodeMetrics:
@@ -533,6 +558,7 @@ def update_from_finished_request(
         max_tokens_param: int | None,
         req_stats: RequestStateStats,
         num_cached_tokens: int = 0,
+        prefill_stats: PrefillStats | None = None,
     ):
         e2e_latency = self._time_since(req_stats.arrival_time)
 
@@ -572,6 +598,7 @@ def update_from_finished_request(
             mean_time_per_output_token=mean_time_per_output_token,
             is_corrupted=req_stats.is_corrupted,
             num_cached_tokens=num_cached_tokens,
+            prefill_stats=prefill_stats,
         )
         self.finished_requests.append(finished_req)
 

From 511ef8b4861da8b7e5ab5e293865586deb941038 Mon Sep 17 00:00:00 2001
From: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
Date: Tue, 25 Aug 2026 17:36:00 -0400
Subject: [PATCH 2/5] docs: warn about prefix-cache telemetry exposure

Document the cross-tenant cache-state exposure of per-request prefix metrics and point operators to per-tenant secret cache salting.

Assisted-by: OpenAI Codex
Signed-off-by: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
---
 docs/features/per_request_metrics.md       | 11 +++++++++++
 vllm/entrypoints/launchers/cli_args.py     |  5 ++++-
 vllm/entrypoints/openai/engine/protocol.py |  8 +++++++-
 3 files changed, 22 insertions(+), 2 deletions(-)

diff --git a/docs/features/per_request_metrics.md b/docs/features/per_request_metrics.md
index bf1e3f2447e0..eb84938d38cd 100644
--- a/docs/features/per_request_metrics.md
+++ b/docs/features/per_request_metrics.md
@@ -16,6 +16,17 @@ vllm serve meta-llama/Llama-3.1-8B-Instruct --enable-per-request-metrics
 When this flag is set, supported API responses include metrics for each
 attributable request.
 
+!!! warning "Security: prefix-cache state"
+    Prefix-cache metrics expose exact per-request cache-hit and eviction state.
+    In a shared deployment, this can reveal whether another tenant populated a
+    guessed prefix without relying on timing analysis. Use this flag only in a
+    trusted single-tenant deployment, or ensure that every request includes an
+    unpredictable secret `cache_salt` scoped to the intended tenant isolation
+    boundary. A shared or predictable salt does not provide cross-tenant
+    isolation. See the
+    [cache-salting guidance](../usage/security.md#prefix-cache-timing-side-channel-mitigation-cache-salting)
+    and [CVE-2025-46570](https://github.com/vllm-project/vllm/security/advisories/GHSA-4qjh-9fv9-r85r).
+
 !!! note
     At high concurrency, enabling per-request metrics computation may introduce
     non-negligible CPU overhead. Benchmark your specific workload to evaluate the
diff --git a/vllm/entrypoints/launchers/cli_args.py b/vllm/entrypoints/launchers/cli_args.py
index 615c65d7fb89..37f4530e8ec0 100644
--- a/vllm/entrypoints/launchers/cli_args.py
+++ b/vllm/entrypoints/launchers/cli_args.py
@@ -132,7 +132,10 @@ class BaseFrontendArgs:
     enable_prompt_tokens_details: bool = False
     """If set to True, enable prompt_tokens_details in usage."""
     enable_per_request_metrics: bool = False
-    """If set to True, include per-request timing metrics in API responses."""
+    """If set to True, include per-request timing and experimental prefix-cache
+    metrics in API responses. Prefix-cache metrics expose exact cache state;
+    review the security guide before enabling them in multi-tenant deployments.
+    """
     enable_server_load_tracking: bool = False
     """If set to True, enable tracking server_load_metrics in the app state."""
     enable_force_include_usage: bool = False
diff --git a/vllm/entrypoints/openai/engine/protocol.py b/vllm/entrypoints/openai/engine/protocol.py
index 58a0a9b1939e..4164bd297b87 100644
--- a/vllm/entrypoints/openai/engine/protocol.py
+++ b/vllm/entrypoints/openai/engine/protocol.py
@@ -152,7 +152,13 @@ class SpeculativeDecodingMetrics(OpenAIBaseModel):
 
 
 class PrefixCacheMetrics(OpenAIBaseModel):
-    """Experimental per-request prompt/KV telemetry."""
+    """Experimental per-request prompt/KV telemetry.
+
+    These fields expose exact cache-hit and eviction state. In multi-tenant
+    deployments, isolate every request with an unpredictable secret
+    ``cache_salt`` scoped to the intended tenant boundary. See the security
+    guide for details.
+    """
 
     num_computed_tokens: int
     num_cached_tokens: int

From 086ef08068909ba8451a44830f4552e30cb315c3 Mon Sep 17 00:00:00 2001
From: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
Date: Wed, 26 Aug 2026 18:31:26 -0400
Subject: [PATCH 3/5] tests: avoid output processor merge conflict

Keep the per-request prefill telemetry test outside the section changed by upstream #53704 and import PrefillStats locally so both changes merge without overlapping import hunks.

Signed-off-by: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
---
 tests/v1/engine/test_output_processor.py | 104 ++++++++++++-----------
 1 file changed, 53 insertions(+), 51 deletions(-)

diff --git a/tests/v1/engine/test_output_processor.py b/tests/v1/engine/test_output_processor.py
index c3161ea0c05e..5863d4b557d6 100644
--- a/tests/v1/engine/test_output_processor.py
+++ b/tests/v1/engine/test_output_processor.py
@@ -28,7 +28,7 @@
     FinishReason,
 )
 from vllm.v1.engine.output_processor import OutputProcessor, RequestOutputCollector
-from vllm.v1.metrics.stats import IterationStats, PrefillStats, SchedulerStats
+from vllm.v1.metrics.stats import IterationStats, SchedulerStats
 
 
 def _ref_convert_id_to_token(
@@ -47,56 +47,6 @@ def _ref_convert_id_to_token(
     return tokenizer.decode([token_id]) or ""
 
 
-def test_prefill_stats_propagate_to_request_output():
-    output_processor = OutputProcessor(tokenizer=None, log_stats=False)
-    request = EngineCoreRequest(
-        request_id="request-internal",
-        external_req_id="request-external",
-        prompt_token_ids=[1, 2, 3, 4],
-        mm_features=None,
-        arrival_time=0,
-        lora_request=None,
-        cache_salt=None,
-        data_parallel_rank=None,
-        sampling_params=SamplingParams(
-            detokenize=False,
-            output_kind=RequestOutputKind.DELTA,
-        ),
-        pooling_params=None,
-    )
-    output_processor.add_request(request, prompt=None)
-
-    prefill_stats = PrefillStats(
-        num_prompt_tokens=4,
-        num_computed_tokens=2,
-        num_cached_tokens=2,
-        num_local_cached_tokens=1,
-        num_external_cached_tokens=1,
-        num_cache_creation_tokens=2,
-        num_new_full_blocks=1,
-        num_block_allocations=2,
-        num_block_evictions=1,
-        num_prefill_chunks=2,
-    )
-    processed = output_processor.process_outputs(
-        [
-            EngineCoreOutput(
-                request_id=request.request_id,
-                new_token_ids=[42],
-                finish_reason=FinishReason.LENGTH,
-                prefill_stats=prefill_stats,
-            )
-        ]
-    )
-
-    assert len(processed.request_outputs) == 1
-    request_output = processed.request_outputs[0]
-    assert isinstance(request_output, RequestOutput)
-    assert request_output.request_id == "request-external"
-    assert request_output.prefill_stats is prefill_stats
-    assert request_output.prefill_stats.num_block_evictions == 1
-
-
 @pytest.mark.parametrize(
     "request_output_kind", [RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY]
 )
@@ -1469,3 +1419,55 @@ def test_abort_requests(runner: str, abort_by: str, dummy_test_vectors):
             output_processor.abort_requests([request.request_id], internal=True)
         else:
             output_processor.abort_requests([request.external_req_id], internal=False)
+
+
+def test_prefill_stats_propagate_to_request_output():
+    from vllm.v1.metrics.stats import PrefillStats
+
+    output_processor = OutputProcessor(tokenizer=None, log_stats=False)
+    request = EngineCoreRequest(
+        request_id="request-internal",
+        external_req_id="request-external",
+        prompt_token_ids=[1, 2, 3, 4],
+        mm_features=None,
+        arrival_time=0,
+        lora_request=None,
+        cache_salt=None,
+        data_parallel_rank=None,
+        sampling_params=SamplingParams(
+            detokenize=False,
+            output_kind=RequestOutputKind.DELTA,
+        ),
+        pooling_params=None,
+    )
+    output_processor.add_request(request, prompt=None)
+
+    prefill_stats = PrefillStats(
+        num_prompt_tokens=4,
+        num_computed_tokens=2,
+        num_cached_tokens=2,
+        num_local_cached_tokens=1,
+        num_external_cached_tokens=1,
+        num_cache_creation_tokens=2,
+        num_new_full_blocks=1,
+        num_block_allocations=2,
+        num_block_evictions=1,
+        num_prefill_chunks=2,
+    )
+    processed = output_processor.process_outputs(
+        [
+            EngineCoreOutput(
+                request_id=request.request_id,
+                new_token_ids=[42],
+                finish_reason=FinishReason.LENGTH,
+                prefill_stats=prefill_stats,
+            )
+        ]
+    )
+
+    assert len(processed.request_outputs) == 1
+    request_output = processed.request_outputs[0]
+    assert isinstance(request_output, RequestOutput)
+    assert request_output.request_id == "request-external"
+    assert request_output.prefill_stats is prefill_stats
+    assert request_output.prefill_stats.num_block_evictions == 1

From 2258331e71ab2b8a3f339f841f5761a2dc2931b6 Mon Sep 17 00:00:00 2001
From: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
Date: Sat, 29 Aug 2026 18:28:26 -0400
Subject: [PATCH 4/5] [Bugfix] Preserve prefill stats across streaming input

Assisted-by: OpenAI Codex <codex@openai.com>
Signed-off-by: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
---
 docs/features/per_request_metrics.md          |  10 +-
 tests/v1/engine/test_output_processor.py      | 114 ++++++++++++++++++
 .../test_scheduler_streaming.py               |  88 +++++++++++++-
 vllm/outputs.py                               |   6 +-
 vllm/v1/core/sched/scheduler.py               |  29 ++++-
 vllm/v1/engine/output_processor.py            |  11 +-
 vllm/v1/metrics/stats.py                      |  30 ++++-
 vllm/v1/request.py                            |   4 +
 8 files changed, 280 insertions(+), 12 deletions(-)

diff --git a/docs/features/per_request_metrics.md b/docs/features/per_request_metrics.md
index eb84938d38cd..9b0c888d3d6e 100644
--- a/docs/features/per_request_metrics.md
+++ b/docs/features/per_request_metrics.md
@@ -85,7 +85,7 @@ prompt and KV-cache telemetry:
 
 | Field | Description |
 | --- | --- |
-| `num_computed_tokens` | Logical prompt tokens assigned to local model computation at first admission. Recomputation after preemption is not double-counted. |
+| `num_computed_tokens` | Logical prompt tokens assigned to local model computation when each input chunk is admitted. Recomputation after preemption is not double-counted. |
 | `num_cached_tokens` | Prompt tokens skipped during local computation (`num_local_cached_tokens + num_external_cached_tokens`). |
 | `num_local_cached_tokens` | Prompt tokens supplied by the local prefix cache. |
 | `num_external_cached_tokens` | Prompt tokens supplied through an external KV transfer. This describes the scheduler source, not whether every transferred block was a cache hit in an upstream deployment. |
@@ -101,6 +101,14 @@ one logical token block can contribute more than one physical block on hybrid
 models. The response's top-level `id` correlates the telemetry with the request;
 the same ID is present on the final streaming chunk.
 
+For streaming-input sessions, prefix-cache telemetry is cumulative across all
+admitted input chunks. Each continuation is measured independently inside the
+scheduler and then added to the prior chunks. Generated tokens retained as the
+next chunk's context are not counted again as prompt input or cache creation;
+physical block activity and prefill-chunk counts still include work attributable
+to every input chunk. Previously emitted `RequestOutput` snapshots remain
+unchanged when a later chunk completes.
+
 These engine-level fields intentionally live under `metrics`, rather than
 OpenAI `usage`: block allocation and eviction are implementation details, and
 external-transfer and local-cache sources are not billing-token categories.
diff --git a/tests/v1/engine/test_output_processor.py b/tests/v1/engine/test_output_processor.py
index 2332bc4523de..b6f8bced80ca 100644
--- a/tests/v1/engine/test_output_processor.py
+++ b/tests/v1/engine/test_output_processor.py
@@ -1501,3 +1501,117 @@ def test_prefill_stats_propagate_to_request_output():
     assert request_output.request_id == "request-external"
     assert request_output.prefill_stats is prefill_stats
     assert request_output.prefill_stats.num_block_evictions == 1
+
+
+def test_streaming_prefill_stats_are_session_cumulative():
+    from vllm.v1.metrics.stats import PrefillStats
+
+    output_processor = OutputProcessor(tokenizer=None, log_stats=False)
+    request = EngineCoreRequest(
+        request_id="request-internal",
+        external_req_id="request-external",
+        prompt_token_ids=[1, 2, 3, 4],
+        mm_features=None,
+        arrival_time=0,
+        lora_request=None,
+        cache_salt=None,
+        data_parallel_rank=None,
+        sampling_params=SamplingParams(
+            detokenize=False,
+            output_kind=RequestOutputKind.DELTA,
+        ),
+        pooling_params=None,
+        resumable=True,
+    )
+    output_processor.add_request(request, prompt=None)
+
+    first_stats = PrefillStats(
+        num_prompt_tokens=4,
+        num_computed_tokens=2,
+        num_cached_tokens=2,
+        num_local_cached_tokens=1,
+        num_external_cached_tokens=1,
+        num_cache_creation_tokens=2,
+        num_new_full_blocks=1,
+        num_block_allocations=2,
+        num_block_evictions=1,
+        num_prefill_chunks=2,
+    )
+    first_processed = output_processor.process_outputs(
+        [
+            EngineCoreOutput(
+                request_id=request.request_id,
+                new_token_ids=[42],
+                finish_reason=FinishReason.LENGTH,
+                prefill_stats=first_stats,
+            )
+        ]
+    )
+    first_output = first_processed.request_outputs[0]
+    assert isinstance(first_output, RequestOutput)
+    assert first_output.prefill_stats is first_stats
+
+    continuation = EngineCoreRequest(
+        request_id=request.request_id,
+        external_req_id=request.external_req_id,
+        prompt_token_ids=[5, 6],
+        mm_features=None,
+        arrival_time=1,
+        lora_request=None,
+        cache_salt=None,
+        data_parallel_rank=None,
+        sampling_params=SamplingParams(
+            detokenize=False,
+            output_kind=RequestOutputKind.DELTA,
+        ),
+        pooling_params=None,
+        resumable=True,
+    )
+    output_processor.add_request(continuation, prompt=None)
+
+    second_stats = PrefillStats(
+        num_prompt_tokens=2,
+        num_computed_tokens=2,
+        num_cache_creation_tokens=1,
+        num_new_full_blocks=1,
+        num_block_allocations=1,
+        num_block_evictions=2,
+        num_prefill_chunks=1,
+    )
+    second_processed = output_processor.process_outputs(
+        [
+            EngineCoreOutput(
+                request_id=request.request_id,
+                new_token_ids=[43],
+                finish_reason=FinishReason.LENGTH,
+                prefill_stats=second_stats,
+            )
+        ]
+    )
+
+    second_output = second_processed.request_outputs[0]
+    assert isinstance(second_output, RequestOutput)
+    cumulative = second_output.prefill_stats
+    assert cumulative is not None
+    assert cumulative is not first_stats
+    assert cumulative is not second_stats
+    assert cumulative.num_prompt_tokens == 6
+    assert cumulative.num_computed_tokens == 4
+    assert cumulative.num_cached_tokens == 2
+    assert cumulative.num_local_cached_tokens == 1
+    assert cumulative.num_external_cached_tokens == 1
+    assert cumulative.num_cache_creation_tokens == 3
+    assert cumulative.num_new_full_blocks == 2
+    assert cumulative.num_block_allocations == 3
+    assert cumulative.num_block_evictions == 3
+    assert cumulative.num_prefill_chunks == 3
+
+    # Replacing the accumulator with a new merged snapshot prevents already
+    # emitted RequestOutputs from changing retroactively.
+    assert first_output.prefill_stats is first_stats
+    assert first_output.prefill_stats.num_prompt_tokens == 4
+    assert first_output.prefill_stats.num_block_evictions == 1
+
+    first_output.add(second_output, aggregate=True)
+    assert first_output.prefill_stats is cumulative
+    assert first_output.prefill_stats.num_prompt_tokens == 6
diff --git a/tests/v1/streaming_input/test_scheduler_streaming.py b/tests/v1/streaming_input/test_scheduler_streaming.py
index 822b4e49da34..1e1711181ae3 100644
--- a/tests/v1/streaming_input/test_scheduler_streaming.py
+++ b/tests/v1/streaming_input/test_scheduler_streaming.py
@@ -13,6 +13,11 @@
     PlaceholderRange,
 )
 from vllm.sampling_params import SamplingParams
+from vllm.utils.hashing import sha256
+from vllm.v1.core.kv_cache_utils import (
+    get_request_block_hasher,
+    init_none_hash,
+)
 from vllm.v1.core.sched.scheduler import Scheduler
 from vllm.v1.engine import FinishReason
 from vllm.v1.kv_cache_interface import (
@@ -35,6 +40,7 @@ def __init__(
         prompt_token_ids=None,
         mm_features: list[MultiModalFeatureSpec] | None = None,
         max_tokens: int | None = 16,
+        block_hasher=None,
     ):
         super().__init__(
             request_id=request_id,
@@ -45,10 +51,11 @@ def __init__(
             pooling_params=None,
             mm_features=mm_features,
             resumable=resumable,
+            block_hasher=block_hasher,
         )
 
 
-def create_scheduler() -> Scheduler:
+def create_scheduler(*, enable_prefix_caching: bool = False) -> Scheduler:
     vllm_config = VllmConfig(device_config=DeviceConfig("cpu"))
     vllm_config.model_config = MagicMock()
     vllm_config.model_config.skip_tokenizer_init = True
@@ -59,7 +66,7 @@ def create_scheduler() -> Scheduler:
     vllm_config.model_config.enable_return_routed_experts = False
     vllm_config.cache_config = MagicMock()
     vllm_config.cache_config.num_gpu_blocks = 1000
-    vllm_config.cache_config.enable_prefix_caching = False
+    vllm_config.cache_config.enable_prefix_caching = enable_prefix_caching
     kv_cache_config = KVCacheConfig(
         num_blocks=1000,
         kv_cache_tensors=[],
@@ -314,6 +321,83 @@ def test_update_request_as_session_with_output_tokens(self):
         num_new_tokens = session.num_tokens - session.num_computed_tokens
         assert num_new_tokens == 2
 
+    def test_two_chunk_prefill_stats_lifecycle(self):
+        """Each streaming chunk is measured and emitted independently.
+
+        The output processor combines these per-chunk snapshots into the
+        session-cumulative telemetry exposed on RequestOutput.
+        """
+        scheduler = create_scheduler(enable_prefix_caching=True)
+        init_none_hash(sha256)
+        session = DummyRequest(
+            request_id="session",
+            prompt_token_ids=list(range(16)),
+            block_hasher=get_request_block_hasher(16, sha256),
+        )
+        scheduler.add_request(session)
+
+        first_schedule = scheduler.schedule()
+        first_outputs = scheduler.update_from_output(
+            first_schedule,
+            ModelRunnerOutput(
+                req_ids=[session.request_id],
+                req_id_to_index={session.request_id: 0},
+                sampled_token_ids=[[STOP_TOKEN]],
+                logprobs=None,
+                prompt_logprobs_dict={session.request_id: None},
+                pooler_output=[],
+            ),
+        )
+        first_stats = first_outputs[session.client_index].outputs[0].prefill_stats
+        assert first_stats is not None
+        assert first_stats.num_prompt_tokens == 16
+        assert first_stats.num_computed_tokens == 16
+        assert first_stats.num_cached_tokens == 0
+        assert first_stats.num_cache_creation_tokens == 16
+        assert first_stats.num_prefill_chunks == 1
+        assert session.prefill_stats is None
+
+        # Continue the same session with enough new input to fill another
+        # cache block. The terminating sampled token from chunk 1 is discarded,
+        # so only these 17 input tokens belong to chunk 2's logical accounting.
+        next_request = DummyRequest(
+            request_id=session.request_id,
+            prompt_token_ids=list(range(100, 117)),
+        )
+        scheduler.add_request(next_request)
+
+        assert session.prefill_stats is not None
+        assert session.prefill_stats.num_prompt_tokens == 17
+        assert session.prefill_stats.num_computed_tokens == 17
+        assert session.prefill_stats.num_cached_tokens == 0
+
+        second_schedule = scheduler.schedule()
+        second_outputs = scheduler.update_from_output(
+            second_schedule,
+            ModelRunnerOutput(
+                req_ids=[session.request_id],
+                req_id_to_index={session.request_id: 0},
+                sampled_token_ids=[[STOP_TOKEN]],
+                logprobs=None,
+                prompt_logprobs_dict={session.request_id: None},
+                pooler_output=[],
+            ),
+        )
+        second_stats = second_outputs[session.client_index].outputs[0].prefill_stats
+        assert second_stats is not None
+        assert second_stats.num_prompt_tokens == 17
+        assert second_stats.num_computed_tokens == 17
+        assert second_stats.num_cached_tokens == 0
+        assert second_stats.num_cache_creation_tokens == 16
+        assert second_stats.num_prefill_chunks == 1
+        assert second_stats.num_new_full_blocks >= 1
+        assert second_stats.num_block_allocations >= 1
+
+        # The first emitted snapshot must not be mutated by the continuation.
+        assert first_stats.num_prompt_tokens == 16
+        assert first_stats.num_cache_creation_tokens == 16
+        assert first_stats.num_prefill_chunks == 1
+
     def test_streaming_e2e_lifecycle(self):
         """
         Comprehensive integration test covering complete streaming request lifecycle
diff --git a/vllm/outputs.py b/vllm/outputs.py
index dbce4c25648f..639102632add 100644
--- a/vllm/outputs.py
+++ b/vllm/outputs.py
@@ -131,7 +131,9 @@ class RequestOutput:
         num_cache_creation_tokens: Prompt tokens currently counted as local
             prefix-cache writes for this request.
         prefill_stats: Per-request prompt/KV telemetry emitted when prefill
-            completes.
+            completes. For streaming input, the latest value is cumulative
+            across admitted input chunks in the same session; retained generated
+            tokens are not counted again as prompt input.
         kv_transfer_params: The params for remote K/V transfer.
         ec_transfer_params: The params for remote encoder-cache transfer.
     """
@@ -184,7 +186,7 @@ def add(self, next_output: "RequestOutput", aggregate: bool) -> None:
         self.finished |= next_output.finished
         self.kv_transfer_params = next_output.kv_transfer_params
         self.ec_transfer_params = next_output.ec_transfer_params
-        if self.prefill_stats is None:
+        if next_output.prefill_stats is not None:
             self.prefill_stats = next_output.prefill_stats
 
         for next_completion in next_output.outputs:
diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py
index c647bb941ec6..da4f6ba8d372 100644
--- a/vllm/v1/core/sched/scheduler.py
+++ b/vllm/v1/core/sched/scheduler.py
@@ -56,6 +56,7 @@
 from vllm.v1.kv_cache_interface import KVCacheConfig, MambaSpec
 from vllm.v1.metrics.perf import ModelMetrics, PerfStats
 from vllm.v1.metrics.stats import (
+    PrefillStats,
     PrefixCacheStats,
     RequestSpecDecodeMetrics,
     SchedulerStats,
@@ -1469,6 +1470,21 @@ def _update_request_as_session(
         Discards the last sampled output token from the prior input chunk.
         """
 
+        # Start a fresh scheduler-side accumulator for this input chunk. The
+        # output processor merges emitted chunk snapshots for the public,
+        # session-cumulative view. Capture the baseline before extending the
+        # prompt so retained session KV is not reported as new cache creation.
+        session.prefill_stats = PrefillStats()
+        session.prefill_stats_cache_baseline = (
+            self.kv_cache_manager.estimate_cached_tokens(session)
+        )
+        new_prompt_tokens = update.prompt_token_ids or ()
+        session.prefill_stats.set(
+            num_prompt_tokens=len(new_prompt_tokens),
+            num_local_cached_tokens=0,
+            num_external_cached_tokens=0,
+        )
+
         # Current streaming input behaviour: Keep only computed output tokens
         # (discard final sampled output token).
         num_computed_tokens = session.num_computed_tokens
@@ -1489,8 +1505,8 @@ def _update_request_as_session(
                 )
             session.mm_features.extend(update.mm_features)
 
-        session._all_token_ids.extend(update.prompt_token_ids or ())
-        session.prompt_token_ids.extend(update.prompt_token_ids or ())
+        session._all_token_ids.extend(new_prompt_tokens)
+        session.prompt_token_ids.extend(new_prompt_tokens)
         # Update block hashes for the new tokens.
         session.update_block_hashes()
         session.num_prompt_tokens = len(session.prompt_token_ids)
@@ -2012,9 +2028,16 @@ def update_from_output(
             if should_emit_output:
                 prefill_stats = request.take_prefill_stats()
                 if prefill_stats is not None:
-                    prefill_stats.finalize(
+                    current_cached_tokens = (
                         self.kv_cache_manager.estimate_cached_tokens(request)
                     )
+                    prefill_stats.finalize(
+                        max(
+                            0,
+                            current_cached_tokens
+                            - request.prefill_stats_cache_baseline,
+                        )
+                    )
 
             finish_reason = None
             if stopped:
diff --git a/vllm/v1/engine/output_processor.py b/vllm/v1/engine/output_processor.py
index 66daf766a3f4..b2350945b917 100644
--- a/vllm/v1/engine/output_processor.py
+++ b/vllm/v1/engine/output_processor.py
@@ -662,12 +662,17 @@ def process_outputs(
 
             if req_state.is_prefilling:
                 if engine_core_output.prefill_stats is not None:
-                    req_state.prefill_stats = engine_core_output.prefill_stats
+                    if req_state.prefill_stats is None:
+                        req_state.prefill_stats = engine_core_output.prefill_stats
+                    else:
+                        req_state.prefill_stats = req_state.prefill_stats.merged(
+                            engine_core_output.prefill_stats
+                        )
                     req_state.num_cached_tokens = (
-                        engine_core_output.prefill_stats.num_cached_tokens
+                        req_state.prefill_stats.num_cached_tokens
                     )
                     req_state.num_cache_creation_tokens = (
-                        engine_core_output.prefill_stats.num_cache_creation_tokens
+                        req_state.prefill_stats.num_cache_creation_tokens
                     )
                 req_state.is_prefilling = False
 
diff --git a/vllm/v1/metrics/stats.py b/vllm/v1/metrics/stats.py
index 6e36c41624ab..7733e7201b3d 100644
--- a/vllm/v1/metrics/stats.py
+++ b/vllm/v1/metrics/stats.py
@@ -258,7 +258,12 @@ class FinishedRequestStats:
 
 @dataclass
 class PrefillStats:
-    """Breakdown of a scheduled prefill computation.
+    """Breakdown of one admitted prompt chunk's prefill computation.
+
+    Streaming-input continuations produce one snapshot per admitted input
+    chunk. The output processor combines those snapshots into a
+    session-cumulative value without counting retained generated tokens as new
+    prompt input.
 
     Fields:
         num_prompt_tokens: Total number of tokens to be prefilled.
@@ -321,6 +326,29 @@ def record_block_activity(
         self.num_block_evictions += num_block_evictions
         self.num_new_full_blocks += num_new_full_blocks
 
+    def merged(self, other: "PrefillStats") -> "PrefillStats":
+        """Return a cumulative snapshot without mutating either operand."""
+        return PrefillStats(
+            num_prompt_tokens=self.num_prompt_tokens + other.num_prompt_tokens,
+            num_computed_tokens=(self.num_computed_tokens + other.num_computed_tokens),
+            num_cached_tokens=self.num_cached_tokens + other.num_cached_tokens,
+            num_local_cached_tokens=(
+                self.num_local_cached_tokens + other.num_local_cached_tokens
+            ),
+            num_external_cached_tokens=(
+                self.num_external_cached_tokens + other.num_external_cached_tokens
+            ),
+            num_cache_creation_tokens=(
+                self.num_cache_creation_tokens + other.num_cache_creation_tokens
+            ),
+            num_new_full_blocks=(self.num_new_full_blocks + other.num_new_full_blocks),
+            num_block_allocations=(
+                self.num_block_allocations + other.num_block_allocations
+            ),
+            num_block_evictions=(self.num_block_evictions + other.num_block_evictions),
+            num_prefill_chunks=self.num_prefill_chunks + other.num_prefill_chunks,
+        )
+
 
 @dataclass
 class RequestSpecDecodeMetrics:
diff --git a/vllm/v1/request.py b/vllm/v1/request.py
index 8b453a09069e..4cf84cefefa2 100644
--- a/vllm/v1/request.py
+++ b/vllm/v1/request.py
@@ -210,6 +210,10 @@ def __init__(
         self.num_preemptions = 0
 
         self.prefill_stats: PrefillStats | None = PrefillStats()
+        # Cached-token estimate at the beginning of the active input chunk.
+        # Streaming continuations retain prior KV, so finalization must use the
+        # increase from this baseline rather than the session-wide total.
+        self.prefill_stats_cache_baseline = 0
 
         # Per-request speculative-decoding acceptance accumulator. Populated by
         # the scheduler when --per-request-spec-decode-metrics is set (eagerly on

From c3032a953f3a20765ba7e518b5cf4f4c6d0abe7f Mon Sep 17 00:00:00 2001
From: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
Date: Sun, 30 Aug 2026 15:25:42 -0400
Subject: [PATCH 5/5] [Bugfix] Exclude retained partial-block KV from
 continuation stats

Assisted-by: OpenAI Codex <codex@openai.com>
Signed-off-by: Guanjie LIN <77091014+cook1e-0707@users.noreply.github.com>
---
 .../test_scheduler_streaming.py               | 42 ++++++++++++++-----
 vllm/v1/core/sched/scheduler.py               | 12 +++---
 vllm/v1/request.py                            |  7 ++--
 3 files changed, 41 insertions(+), 20 deletions(-)

diff --git a/tests/v1/streaming_input/test_scheduler_streaming.py b/tests/v1/streaming_input/test_scheduler_streaming.py
index 1e1711181ae3..e5c18150d096 100644
--- a/tests/v1/streaming_input/test_scheduler_streaming.py
+++ b/tests/v1/streaming_input/test_scheduler_streaming.py
@@ -342,7 +342,7 @@ def test_two_chunk_prefill_stats_lifecycle(self):
             ModelRunnerOutput(
                 req_ids=[session.request_id],
                 req_id_to_index={session.request_id: 0},
-                sampled_token_ids=[[STOP_TOKEN]],
+                sampled_token_ids=[[1000]],
                 logprobs=None,
                 prompt_logprobs_dict={session.request_id: None},
                 pooler_output=[],
@@ -357,18 +357,40 @@ def test_two_chunk_prefill_stats_lifecycle(self):
         assert first_stats.num_prefill_chunks == 1
         assert session.prefill_stats is None
 
-        # Continue the same session with enough new input to fill another
-        # cache block. The terminating sampled token from chunk 1 is discarded,
-        # so only these 17 input tokens belong to chunk 2's logical accounting.
+        # Compute one generated token, then sample a terminating token. The
+        # generated token retains valid KV in the next partial cache block;
+        # only the uncomputed terminating token is discarded at continuation.
+        decode_schedule = scheduler.schedule()
+        scheduler.update_from_output(
+            decode_schedule,
+            ModelRunnerOutput(
+                req_ids=[session.request_id],
+                req_id_to_index={session.request_id: 0},
+                sampled_token_ids=[[STOP_TOKEN]],
+                logprobs=None,
+                prompt_logprobs_dict={session.request_id: None},
+                pooler_output=[],
+            ),
+        )
+        assert session.num_computed_tokens == 17
+
+        # Add exactly one cache block of new input. The block ending at token
+        # 32 contains one retained generated token and 15 new prompt tokens;
+        # the sixteenth new token remains in the trailing partial block.
         next_request = DummyRequest(
             request_id=session.request_id,
-            prompt_token_ids=list(range(100, 117)),
+            prompt_token_ids=list(range(100, 116)),
         )
         scheduler.add_request(next_request)
 
+        assert session.prompt_token_ids == [
+            *range(16),
+            1000,
+            *range(100, 116),
+        ]
         assert session.prefill_stats is not None
-        assert session.prefill_stats.num_prompt_tokens == 17
-        assert session.prefill_stats.num_computed_tokens == 17
+        assert session.prefill_stats.num_prompt_tokens == 16
+        assert session.prefill_stats.num_computed_tokens == 16
         assert session.prefill_stats.num_cached_tokens == 0
 
         second_schedule = scheduler.schedule()
@@ -385,10 +407,10 @@ def test_two_chunk_prefill_stats_lifecycle(self):
         )
         second_stats = second_outputs[session.client_index].outputs[0].prefill_stats
         assert second_stats is not None
-        assert second_stats.num_prompt_tokens == 17
-        assert second_stats.num_computed_tokens == 17
+        assert second_stats.num_prompt_tokens == 16
+        assert second_stats.num_computed_tokens == 16
         assert second_stats.num_cached_tokens == 0
-        assert second_stats.num_cache_creation_tokens == 16
+        assert second_stats.num_cache_creation_tokens == 15
         assert second_stats.num_prefill_chunks == 1
         assert second_stats.num_new_full_blocks >= 1
         assert second_stats.num_block_allocations >= 1
diff --git a/vllm/v1/core/sched/scheduler.py b/vllm/v1/core/sched/scheduler.py
index da4f6ba8d372..c10f9d1aecdf 100644
--- a/vllm/v1/core/sched/scheduler.py
+++ b/vllm/v1/core/sched/scheduler.py
@@ -1472,12 +1472,12 @@ def _update_request_as_session(
 
         # Start a fresh scheduler-side accumulator for this input chunk. The
         # output processor merges emitted chunk snapshots for the public,
-        # session-cumulative view. Capture the baseline before extending the
-        # prompt so retained session KV is not reported as new cache creation.
+        # session-cumulative view. The continuation starts after every token
+        # with valid retained KV, including generated tokens in a partial
+        # cache block, so none of that prior state is reported as new cache
+        # creation.
         session.prefill_stats = PrefillStats()
-        session.prefill_stats_cache_baseline = (
-            self.kv_cache_manager.estimate_cached_tokens(session)
-        )
+        session.prefill_stats_cache_creation_start = session.num_computed_tokens
         new_prompt_tokens = update.prompt_token_ids or ()
         session.prefill_stats.set(
             num_prompt_tokens=len(new_prompt_tokens),
@@ -2035,7 +2035,7 @@ def update_from_output(
                         max(
                             0,
                             current_cached_tokens
-                            - request.prefill_stats_cache_baseline,
+                            - request.prefill_stats_cache_creation_start,
                         )
                     )
 
diff --git a/vllm/v1/request.py b/vllm/v1/request.py
index 4cf84cefefa2..ad956fa7c653 100644
--- a/vllm/v1/request.py
+++ b/vllm/v1/request.py
@@ -210,10 +210,9 @@ def __init__(
         self.num_preemptions = 0
 
         self.prefill_stats: PrefillStats | None = PrefillStats()
-        # Cached-token estimate at the beginning of the active input chunk.
-        # Streaming continuations retain prior KV, so finalization must use the
-        # increase from this baseline rather than the session-wide total.
-        self.prefill_stats_cache_baseline = 0
+        # Token position where the active input chunk begins. Cache-creation
+        # accounting excludes retained session KV before this boundary.
+        self.prefill_stats_cache_creation_start = 0
 
         # Per-request speculative-decoding acceptance accumulator. Populated by
         # the scheduler when --per-request-spec-decode-metrics is set (eagerly on