feat(observability): report real prefix-cache query/hit counters in /metrics

pegainfer-frontend/src/engine/metrics.rs

@@ -26,6 +26,12 @@ pub struct SchedulerMetrics {
     pub num_waiting_reqs: u64,
     /// Cumulative spec-decode counters, or `None` when no draft model is loaded.
     pub spec_decode: Option<SpecDecodeCounters>,
+    /// Cumulative prefix-cache queries (one per request that reached its first
+    /// prefill chunk). Monotonic; mapped to vLLM `PrefixCacheStats.queries`.
+    pub prefix_cache_queries: u64,
+    /// Cumulative prefix-cache hits, in tokens. Monotonic; mapped to vLLM
+    /// `PrefixCacheStats.hits`.
+    pub prefix_cache_hits: u64,
 }
 
 /// Upper bound on a drafter's `K`, fixing the width of

pegainfer-frontend/src/vllm/bridge.rs

@@ -35,7 +35,9 @@ use vllm_engine_core_client::protocol::output::StopReason;
 use vllm_engine_core_client::protocol::output::UtilityCallOutput;
 use vllm_engine_core_client::protocol::request::EngineCoreRequest;
 use vllm_engine_core_client::protocol::request::EngineCoreRequestType;
+use vllm_engine_core_client::protocol::stats::BaseCacheStats;
 use vllm_engine_core_client::protocol::stats::PrefillStats;
+use vllm_engine_core_client::protocol::stats::PrefixCacheStats;
 use vllm_engine_core_client::protocol::stats::SchedulerStats;
 use vllm_engine_core_client::protocol::stats::SpecDecodingStats;
 use vllm_engine_core_client::protocol::utility::UtilityCallId;
@@ -597,6 +599,14 @@ pub(crate) fn scheduler_stats_from(snapshot: &SchedulerMetrics) -> SchedulerStat
         } else {
             snapshot.kv_used_blocks as f64 / snapshot.kv_total_blocks as f64
         },
+        prefix_cache_stats: PrefixCacheStats {
+            base: BaseCacheStats {
+                queries: snapshot.prefix_cache_queries,
+                hits: snapshot.prefix_cache_hits,
+                ..BaseCacheStats::default()
+            },
+            ..PrefixCacheStats::default()
+        },
         ..SchedulerStats::default()
     }
 }

pegainfer-frontend/src/vllm/bridge/tests.rs

@@ -570,6 +570,8 @@ async fn load_snapshots_become_stats_only_batches() {
         num_running_reqs: 2,
         num_waiting_reqs: 1,
         spec_decode: None,
+        prefix_cache_queries: 0,
+        prefix_cache_hits: 0,
     });
     let (output_tx, mut output_rx) = mpsc::unbounded_channel();
     let shutdown = CancellationToken::new();

pegainfer-qwen3/src/frontend_adapter.rs

@@ -241,6 +241,12 @@ pub(crate) struct Qwen3Scheduler<E: ModelExecutor> {
     /// it cannot run against an adapter set the command was about to change.
     pending_control: VecDeque<LoraControl>,
     post_control_deferred: Vec<PendingRequest>,
+    /// Cumulative prefix-cache queries (one per admitted request that reached
+    /// its first prefill chunk). Monotonic; reported verbatim in `SchedulerMetrics`.
+    prefix_cache_queries: u64,
+    /// Cumulative prefix-cache hits, in tokens (sum of cached prefix lengths).
+    /// Monotonic; reported verbatim in `SchedulerMetrics`.
+    prefix_cache_hits: u64,
 }
 
 impl<E: ModelExecutor> Qwen3Scheduler<E> {
@@ -265,6 +271,8 @@ impl<E: ModelExecutor> Qwen3Scheduler<E> {
             lora_rx,
             pending_control: VecDeque::new(),
             post_control_deferred: Vec::new(),
+            prefix_cache_queries: 0,
+            prefix_cache_hits: 0,
         }
     }
 
@@ -342,6 +350,11 @@ impl<E: ModelExecutor> Qwen3Scheduler<E> {
         // terminal rides the committed step, which the driver ships after
         // publishing metrics — the finishing batch's send-time stats then
         // read the drained occupancy instead of racing the publish.
+        // Fold this step's prefix-cache counters into the cumulative totals
+        // before any effect is dropped, so retries/re-queues never lose a count.
+        self.prefix_cache_queries += effects.prefix_queries;
+        self.prefix_cache_hits += effects.prefix_hits;
+
         let mut finishes: Vec<(RequestId, FinishReason)> = Vec::new();
 
         for cached in effects.cached {
@@ -801,6 +814,8 @@ impl<E: ModelExecutor> Scheduler for Qwen3Scheduler<E> {
                 + self.loading.len()
                 + self.post_control_deferred.len()) as u64,
             spec_decode: self.executor.spec_decode_counters(),
+            prefix_cache_queries: self.prefix_cache_queries,
+            prefix_cache_hits: self.prefix_cache_hits,
         }
     }
 }

pegainfer-qwen3/src/scheduler/effects.rs

@@ -87,6 +87,14 @@ pub(crate) struct StepEffects {
     pub(crate) prompt_echoes: Vec<PromptEchoEffect>,
     pub(crate) pending: Vec<PendingEffect>,
     pub(crate) decode: Vec<DecodeEffect>,
+    /// Prefix-cache queries counted this step (one per request whose first
+    /// prefill chunk was resolved — that is where the cache is consulted).
+    /// Cumulative counters live on the scheduler; this is the per-step delta.
+    pub(crate) prefix_queries: u64,
+    /// Prefix-cache hits counted this step, in tokens (the cached prefix
+    /// length of each first-chunk request). Cumulative counters live on the
+    /// scheduler; this is the per-step delta.
+    pub(crate) prefix_hits: u64,
 }
 
 impl StepEffects {
@@ -96,6 +104,8 @@ impl StepEffects {
             prompt_echoes: Vec::new(),
             pending: Vec::new(),
             decode: Vec::new(),
+            prefix_queries: 0,
+            prefix_hits: 0,
         }
     }
 }

pegainfer-qwen3/src/scheduler/resolve.rs

@@ -27,12 +27,16 @@ pub(crate) fn resolve_step(
             prompt_echoes: Vec::new(),
             pending: Vec::new(),
             decode: resolve_decode_outputs(executor, active, &result.requests),
+            prefix_queries: 0,
+            prefix_hits: 0,
         },
         ExecutionArtifacts::SpeculativeDecode { verify } => StepEffects {
             cached: Vec::new(),
             prompt_echoes: Vec::new(),
             pending: Vec::new(),
             decode: resolve_speculative_outputs(executor, active, &verify.requests),
+            prefix_queries: 0,
+            prefix_hits: 0,
         },
         ExecutionArtifacts::Unified { pending, result } => {
             let mut effects = resolve_prefill_outputs(executor, pending, result.prefill_requests);
@@ -107,6 +111,10 @@ fn resolve_prefill_outputs(
                 request_id: req.request_id,
                 cached_tokens: result.cached_tokens,
             });
+            // One cache query per request (the cache is consulted once, on the
+            // first chunk); the cached prefix length is the hit count, in tokens.
+            effects.prefix_queries += 1;
+            effects.prefix_hits += result.cached_tokens as u64;
         }
 
         if !result.completed {