diff --git a/pegainfer-frontend/src/engine/metrics.rs b/pegainfer-frontend/src/engine/metrics.rs index 8fead9060..1656522e0 100644 --- a/pegainfer-frontend/src/engine/metrics.rs +++ b/pegainfer-frontend/src/engine/metrics.rs @@ -26,6 +26,15 @@ pub struct SchedulerMetrics { pub num_waiting_reqs: u64, /// Cumulative spec-decode counters, or `None` when no draft model is loaded. pub spec_decode: Option, + /// 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, token-granularity (the total number of + /// queried prompt tokens already cached). Same unit as + /// `prefix_cache_queries`, so `hit_rate = hits/queries` stays in [0, 1]. + /// Monotonic; mapped to vLLM `PrefixCacheStats.hits` as a per-send delta + /// (see `scheduler_stats_from`), never as the running total. + pub prefix_cache_hits: u64, } /// Upper bound on a drafter's `K`, fixing the width of diff --git a/pegainfer-frontend/src/vllm/bridge.rs b/pegainfer-frontend/src/vllm/bridge.rs index 4b851f3a6..ca7d6e333 100644 --- a/pegainfer-frontend/src/vllm/bridge.rs +++ b/pegainfer-frontend/src/vllm/bridge.rs @@ -35,6 +35,7 @@ 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::SchedulerStats; use vllm_engine_core_client::protocol::stats::SpecDecodingStats; @@ -589,6 +590,14 @@ fn stop_sentinel_id(eos_token_id: Option, stop_token_ids: &[u32]) -> Option /// vLLM `SchedulerStats` view of a load snapshot — what the frontend's /// Prometheus gauges (`scheduler_running`, `scheduler_waiting`, /// `kv_cache_usage`) and DP load balancer consume. +/// +/// `prefix_cache_stats` is left at zero here on purpose: the running totals in +/// `SchedulerMetrics` must NOT be shipped as-is, because the frontend's +/// Prometheus logger increments its `prefix_cache_*_total` counters by the +/// value of *every* `SchedulerStats` it receives. The bridge therefore exports +/// per-send **deltas** (see [`PrefixCacheTracker`]) so a cached request does +/// not re-add the whole history on each subsequent token batch. Callers fill +/// `prefix_cache_stats` from [`PrefixCacheTracker::interval`] after this call. pub(crate) fn scheduler_stats_from(snapshot: &SchedulerMetrics) -> SchedulerStats { SchedulerStats { num_running_reqs: snapshot.num_running_reqs, @@ -602,6 +611,36 @@ pub(crate) fn scheduler_stats_from(snapshot: &SchedulerMetrics) -> SchedulerStat } } +/// Prefix-cache counterpart of [`SpecDecodeTracker`]: the scheduler holds +/// running totals while the wire must carry per-interval deltas, because the +/// frontend increments its `prefix_cache_*_total` counters by the value of +/// *every* `SchedulerStats` it receives. Shipping the running total would +/// re-add the whole history on each subsequent batch, so both bridges convert +/// through this type and cannot drift. +#[derive(Default)] +pub(crate) struct PrefixCacheTracker { + last_queries: u64, + last_hits: u64, +} + +impl PrefixCacheTracker { + /// The delta to stamp on the next outgoing batch. Advances the baseline, so + /// a caller that declines to send after calling this drops only a no-op + /// interval. + pub(crate) fn interval(&mut self, snapshot: &SchedulerMetrics) -> BaseCacheStats { + let delta = BaseCacheStats { + queries: snapshot + .prefix_cache_queries + .saturating_sub(self.last_queries), + hits: snapshot.prefix_cache_hits.saturating_sub(self.last_hits), + ..BaseCacheStats::default() + }; + self.last_queries = snapshot.prefix_cache_queries; + self.last_hits = snapshot.prefix_cache_hits; + delta + } +} + /// Per-interval spec-decode delta from two cumulative snapshots, in the wire /// shape the frontend increments its `vllm:spec_decode_*_total` counters by (see /// [`SpecDecodeCounters`] for why the transport carries totals and the wire @@ -662,9 +701,11 @@ async fn publish_scheduler_stats( shutdown: CancellationToken, ) -> Result<()> { let mut spec = SpecDecodeTracker::default(); + let mut prefix = PrefixCacheTracker::default(); loop { let snapshot = *load_rx.borrow_and_update(); let mut stats = scheduler_stats_from(&snapshot); + stats.prefix_cache_stats.base = prefix.interval(&snapshot); stats.spec_decoding_stats = spec.interval(&snapshot); let outputs = RequestBatchOutputs { engine_index, diff --git a/pegainfer-frontend/src/vllm/bridge/stepped.rs b/pegainfer-frontend/src/vllm/bridge/stepped.rs index 0a1d037cc..62604f68e 100644 --- a/pegainfer-frontend/src/vllm/bridge/stepped.rs +++ b/pegainfer-frontend/src/vllm/bridge/stepped.rs @@ -38,6 +38,7 @@ use zeromq::ZmqMessage; use zeromq::prelude::SocketRecv; use super::BridgeLink; +use super::PrefixCacheTracker; use super::SpecDecodeTracker; use super::connect_link; use super::engine_output; @@ -79,6 +80,7 @@ impl SteppedEngineBridge { .take_steps() .context("partition step stream already taken")?; let mut spec = SpecDecodeTracker::default(); + let mut prefix = PrefixCacheTracker::default(); // Stats are pull-at-send: no push task, the load cell is read when a // batch goes out (and once here, so the frontend's gauges initialize // before any traffic). An idle engine publishes nothing. @@ -97,11 +99,14 @@ impl SteppedEngineBridge { &shutdown, ) .await?; + // Seed the gauges before any traffic. Ship the delta since last send + // (zero here, the first interval) so the frontend's *_total counters + // accumulate deltas, never the running total. send_outputs( &output_tx, RequestBatchOutputs { engine_index: self.engine_index, - scheduler_stats: Some(Box::new(self.stats(&mut spec))), + scheduler_stats: Some(Box::new(self.stats(&mut prefix, &mut spec))), timestamp: now_secs_f64(), ..Default::default() } @@ -144,6 +149,7 @@ impl SteppedEngineBridge { &anchor, &mut streams, &mut names, + &mut prefix, &mut spec, &output_tx, ) { @@ -182,9 +188,14 @@ impl SteppedEngineBridge { /// Stats for an outgoing batch; the spec delta runs from the last batch /// stamped, not the last step run. - fn stats(&self, spec: &mut SpecDecodeTracker) -> SchedulerStats { + fn stats( + &self, + prefix: &mut PrefixCacheTracker, + spec: &mut SpecDecodeTracker, + ) -> SchedulerStats { let snapshot = self.scheduler.metrics(); let mut stats = scheduler_stats_from(&snapshot); + stats.prefix_cache_stats.base = prefix.interval(&snapshot); stats.spec_decoding_stats = spec.interval(&snapshot); stats } @@ -195,6 +206,7 @@ impl SteppedEngineBridge { anchor: &UnixAnchor, streams: &mut HashMap, names: &mut HashMap, + prefix: &mut PrefixCacheTracker, spec: &mut SpecDecodeTracker, output_tx: &tokio::sync::mpsc::UnboundedSender< vllm_engine_core_client::protocol::output::EngineCoreOutputs, @@ -225,7 +237,7 @@ impl SteppedEngineBridge { if outputs.is_empty() { // A drafted step with no batch to ride would strand its increment // until the next batch, which may never come. - let stats = self.stats(spec); + let stats = self.stats(prefix, spec); if stats.spec_decoding_stats.is_some() { send_outputs( output_tx, @@ -244,13 +256,16 @@ impl SteppedEngineBridge { // load before committing the step), so the batch carries stats that // match its own tokens — a finishing batch reports the drained state // and the gauges settle instead of freezing at the last busy value. + // Ship only the delta since the last send so the frontend's + // `prefix_cache_*_total` counters accumulate increments, not the whole + // running total on every batch. send_outputs( output_tx, RequestBatchOutputs { engine_index: self.engine_index, outputs, finished_requests: (!finished_requests.is_empty()).then_some(finished_requests), - scheduler_stats: Some(Box::new(self.stats(spec))), + scheduler_stats: Some(Box::new(self.stats(prefix, spec))), timestamp: now_secs_f64(), } .into(), diff --git a/pegainfer-frontend/src/vllm/bridge/tests.rs b/pegainfer-frontend/src/vllm/bridge/tests.rs index 0926609f7..e39747a30 100644 --- a/pegainfer-frontend/src/vllm/bridge/tests.rs +++ b/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(); @@ -679,3 +681,59 @@ async fn spec_stats_are_per_interval_deltas_that_skip_idle_intervals() { .expect("stats task exits on shutdown") .expect("stats publisher shuts down cleanly"); } + +/// The scheduler holds running prefix-cache totals; the bridge must ship +/// per-send DELTAS (like spec-decode), never the running total. Otherwise the +/// frontend's `prefix_cache_*_total` counters re-add the whole history on every +/// token batch and overcount until restart. Regression test for the P2 review +/// on the prefix-cache counters. +#[tokio::test] +async fn prefix_cache_stats_are_per_interval_deltas_not_running_totals() { + let (load_tx, load_rx) = tokio::sync::watch::channel(SchedulerMetrics { + prefix_cache_queries: 100, + prefix_cache_hits: 37, + ..SchedulerMetrics::default() + }); + let (output_tx, mut output_rx) = mpsc::unbounded_channel(); + let shutdown = CancellationToken::new(); + let task = tokio::spawn(publish_scheduler_stats( + 0, + load_rx, + output_tx, + shutdown.clone(), + )); + + // First publish diffs against zero: carries the whole cumulative. + let first = next_scheduler_stats(&mut output_rx).await; + let pc = first.prefix_cache_stats.base; + assert_eq!(pc.queries, 100, "first interval diffs against zero"); + assert_eq!(pc.hits, 37); + + // The running total doubles; the sent value must be the DELTA (100, 37), + // not the new running total (200, 74). This is the overcount bug. + load_tx.send_replace(SchedulerMetrics { + prefix_cache_queries: 200, + prefix_cache_hits: 74, + ..SchedulerMetrics::default() + }); + let second = next_scheduler_stats(&mut output_rx).await; + let pc = second.prefix_cache_stats.base; + assert_eq!(pc.queries, 100, "second interval ships the delta, not 200"); + assert_eq!(pc.hits, 37, "second interval ships the delta, not 74"); + + // No further change: an idle interval ships a zero delta (not a re-add). + load_tx.send_replace(SchedulerMetrics { + prefix_cache_queries: 200, + prefix_cache_hits: 74, + ..SchedulerMetrics::default() + }); + let idle = next_scheduler_stats(&mut output_rx).await; + let pc = idle.prefix_cache_stats.base; + assert_eq!(pc.queries, 0, "unchanged interval ships a zero delta"); + assert_eq!(pc.hits, 0); + + shutdown.cancel(); + task.await + .expect("stats task exits on shutdown") + .expect("stats publisher shuts down cleanly"); +} diff --git a/pegainfer-qwen3/src/frontend_adapter.rs b/pegainfer-qwen3/src/frontend_adapter.rs index e02125ea5..ab1ccf33b 100644 --- a/pegainfer-qwen3/src/frontend_adapter.rs +++ b/pegainfer-qwen3/src/frontend_adapter.rs @@ -242,6 +242,16 @@ pub(crate) struct Qwen3Scheduler { /// it cannot run against an adapter set the command was about to change. pending_control: VecDeque, post_control_deferred: Vec, + /// 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, token-granularity (the total number of + /// queried prompt tokens that were already cached, summed across requests). + /// Same unit as `prefix_cache_queries`, so `hit_rate = hits/queries` ∈ [0, 1]. + /// Monotonic; the bridge exports per-send deltas of this total (see + /// `scheduler_stats_from` / the stepped bridge) so Prometheus does not + /// re-add the running total on every token batch. + prefix_cache_hits: u64, } impl Qwen3Scheduler { @@ -266,6 +276,8 @@ impl Qwen3Scheduler { lora_rx, pending_control: VecDeque::new(), post_control_deferred: Vec::new(), + prefix_cache_queries: 0, + prefix_cache_hits: 0, } } @@ -343,6 +355,11 @@ impl Qwen3Scheduler { // 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 { @@ -802,6 +819,8 @@ impl Scheduler for Qwen3Scheduler { + 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, } } } diff --git a/pegainfer-qwen3/src/frontend_adapter/tests.rs b/pegainfer-qwen3/src/frontend_adapter/tests.rs index 35441bc7d..f1e442324 100644 --- a/pegainfer-qwen3/src/frontend_adapter/tests.rs +++ b/pegainfer-qwen3/src/frontend_adapter/tests.rs @@ -371,3 +371,119 @@ fn lora_control_waits_until_scheduler_idle() { .expect_err("adapter load should be a stub error"); assert!(matches!(error, LoraControlError::Failed(_))); } + +/// E2E (no GPU) for the prefix-cache `/metrics` counters: drive a real +/// `Qwen3Scheduler` through the engine contract with a fake executor that +/// reports a cached prefix on every request's first chunk, then scrape +/// `SchedulerMetrics` across multiple batches and scrapes. +/// +/// Guards against the two bugs from review: +/// * unit mismatch — `prefix_queries`/`prefix_hits` are both TOKEN-granular +/// (queried prompt tokens / cached tokens), matching vLLM's +/// `PrefixCacheStats`, so `hit_rate = hits/queries` stays in [0, 1]; +/// * cumulative-counter double counting — every request is counted exactly +/// once (on its first prefill chunk), and the bridge exports per-send +/// deltas of the running total (see `prefix_cache_delta`), so repeated +/// scrapes at the same instant are identical and the frontend does not +/// re-add history on every token batch. +#[test] +fn prefix_cache_metrics_stable_across_batches_and_scrapes() { + const BATCHES: u64 = 4; + const PER_BATCH: u64 = 3; + const TOTAL: u64 = BATCHES * PER_BATCH; + const PROMPT_TOKENS: u64 = 64; // tokens looked up per request + const HIT_TOKENS: u64 = 37; // simulated cached prefix length (<= prompt) + + // A fake KV cache that reports a 37-token hit on the first chunk of every + // request. With the fix, queries count the 64 prompt tokens queried and + // hits count the 37 tokens already cached — both token-granular. + let dropped = Arc::new(Mutex::new(Vec::new())); + let executor = FakeExecutor::new(64, Arc::clone(&dropped)).with_prefix_hit(HIT_TOKENS as usize); + let (partition, _lora, mut steps) = launch(executor, false); + + let mut controls = Vec::new(); + let mut scrapes: Vec<(u64, u64, u64)> = Vec::new(); // (batch, queries, hits) + + for batch in 0..BATCHES { + for _ in 0..PER_BATCH { + controls.push(partition.handle.submit(request(PROMPT_TOKENS as usize, 4))); + } + // Wait until this batch's prefills have been counted. + let target = (batch + 1) * PER_BATCH * PROMPT_TOKENS; + assert!( + wait_until(Duration::from_secs(2), || partition + .handle + .metrics() + .prefix_cache_queries + >= target), + "batch {batch} prefix queries never reached {target}" + ); + let m = partition.handle.metrics(); + eprintln!( + "[scrape] after batch {batch}: prefix_cache_queries={} prefix_cache_hits={}", + m.prefix_cache_queries, m.prefix_cache_hits + ); + scrapes.push((batch, m.prefix_cache_queries, m.prefix_cache_hits)); + } + + // Drain the step stream so the driver thread can exit cleanly. + for c in &controls { + let _ = steps.collect_terminal(c.id()); + } + + // Final stable snapshot: repeated scrapes at the same instant are identical. + let a = partition.handle.metrics(); + let b = partition.handle.metrics(); + let c = partition.handle.metrics(); + assert_eq!(a.prefix_cache_queries, b.prefix_cache_queries); + assert_eq!(a.prefix_cache_hits, b.prefix_cache_hits); + assert_eq!(b.prefix_cache_queries, c.prefix_cache_queries); + assert_eq!(b.prefix_cache_hits, c.prefix_cache_hits); + eprintln!( + "[scrape] final (x3 identical): prefix_cache_queries={} prefix_cache_hits={}", + a.prefix_cache_queries, a.prefix_cache_hits + ); + + // Token-granular correctness per vLLM PrefixCacheStats: every request + // queries PROMPT_TOKENS and hits HIT_TOKENS, so the running totals are + // TOTAL * those, and hit_rate = HIT_TOKENS / PROMPT_TOKENS in [0, 1]. + let expected_q = TOTAL * PROMPT_TOKENS; + let expected_h = TOTAL * HIT_TOKENS; + assert_eq!(a.prefix_cache_queries, expected_q); + assert_eq!(a.prefix_cache_hits, expected_h); + assert!( + a.prefix_cache_hits <= a.prefix_cache_queries, + "hits (cached tokens) must not exceed queries (queried tokens)" + ); + let hit_rate = a.prefix_cache_hits as f64 / a.prefix_cache_queries as f64; + eprintln!( + "[rate] prefix hit_rate={:.3} (== {}/{})", + hit_rate, HIT_TOKENS, PROMPT_TOKENS + ); + assert!((hit_rate - HIT_TOKENS as f64 / PROMPT_TOKENS as f64).abs() < 1e-9); + + // Each batch contributed a stable, non-zero delta of exactly + // PER_BATCH * PROMPT_TOKENS (queries) and PER_BATCH * HIT_TOKENS (hits). + let mut prev_q = 0u64; + let mut prev_h = 0u64; + for (batch, q, h) in scrapes { + let dq = q - prev_q; + let dh = h - prev_h; + eprintln!( + "[delta] batch {batch}: +queries={dq} +hits={dh} (hit_rate={:.3})", + h as f64 / q.max(1) as f64 + ); + assert_eq!( + dq, + PER_BATCH * PROMPT_TOKENS, + "batch {batch} added exactly PER_BATCH*PROMPT_TOKENS queries" + ); + assert_eq!( + dh, + PER_BATCH * HIT_TOKENS, + "batch {batch} added exactly PER_BATCH*HIT_TOKENS hits" + ); + prev_q = q; + prev_h = h; + } +} diff --git a/pegainfer-qwen3/src/scheduler/effects.rs b/pegainfer-qwen3/src/scheduler/effects.rs index 8e37a70e3..e2f89a506 100644 --- a/pegainfer-qwen3/src/scheduler/effects.rs +++ b/pegainfer-qwen3/src/scheduler/effects.rs @@ -87,6 +87,17 @@ pub(crate) struct StepEffects { pub(crate) prompt_echoes: Vec, pub(crate) pending: Vec, pub(crate) decode: Vec, + /// Prefix-cache queries counted this step, token-granularity: the number of + /// prompt tokens looked up in the cache (summed over requests whose first + /// prefill chunk was resolved this step — that is where the cache is + /// consulted). Same unit as `prefix_hits`. Cumulative counters live on the + /// scheduler; this is the per-step delta. + pub(crate) prefix_queries: u64, + /// Prefix-cache hits counted this step, token-granularity: the number of + /// queried prompt tokens that were already cached (`cached_tokens`). Same + /// unit as `prefix_queries`, so `hit_rate = hits/queries` stays in [0, 1]. + /// Cumulative counters live on the scheduler; this is the per-step delta. + pub(crate) prefix_hits: u64, } impl StepEffects { @@ -96,6 +107,8 @@ impl StepEffects { prompt_echoes: Vec::new(), pending: Vec::new(), decode: Vec::new(), + prefix_queries: 0, + prefix_hits: 0, } } } diff --git a/pegainfer-qwen3/src/scheduler/resolve.rs b/pegainfer-qwen3/src/scheduler/resolve.rs index 12356210f..8cea807cd 100644 --- a/pegainfer-qwen3/src/scheduler/resolve.rs +++ b/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); @@ -100,13 +104,24 @@ fn resolve_prefill_outputs( // release builds too. assert_eq!(req.request_id, result.request_id); - // Report the prefix-cache hit count on the request's first chunk only - // — that is where it is determined. Later chunks must not re-report. + // Report the prefix-cache counters on the request's first chunk only — + // that is where they are determined. Later chunks must not re-report. + // + // Both counters are TOKEN-granularity, matching vLLM's `PrefixCacheStats` + // (the frontend compares `hits / queries` as hit tokens / queried tokens): + // * `prefix_queries` = the number of prompt tokens this request looked + // up in the cache (the whole prompt is consulted once, on chunk 0). + // * `prefix_hits` = the number of those tokens already cached + // (`cached_tokens`). + // Because `cached_tokens <= prompt_tokens`, `hits <= queries` holds and + // the hit rate stays in [0, 1] with no impossible >100% rates. if req.prefill_pos == 0 { effects.cached.push(CachedTokensEffect { request_id: req.request_id, cached_tokens: result.cached_tokens, }); + effects.prefix_queries += req.prompt_tokens.len() as u64; + effects.prefix_hits += result.cached_tokens as u64; } if !result.completed { diff --git a/pegainfer-qwen3/src/scheduler/test_support.rs b/pegainfer-qwen3/src/scheduler/test_support.rs index 16d19323a..98f6b833d 100644 --- a/pegainfer-qwen3/src/scheduler/test_support.rs +++ b/pegainfer-qwen3/src/scheduler/test_support.rs @@ -39,6 +39,10 @@ pub(crate) struct FakeExecutor { pub(crate) dropped: Arc>>, pub(crate) prefetch_offers: Arc>>, stop_token: Option, + // When > 0, the first prefill chunk of every request reports this many + // `cached_tokens` (a simulated prefix-cache hit). Drives the prefix-cache + // query/hit counters without a real GPU KV cache. + prefix_hit_tokens: usize, } impl FakeExecutor { @@ -56,9 +60,17 @@ impl FakeExecutor { dropped, prefetch_offers: Arc::new(Mutex::new(Vec::new())), stop_token: None, + prefix_hit_tokens: 0, } } + /// Simulate a prefix-cache hit on every request's first prefill chunk by + /// reporting `tokens` cached tokens. + pub(crate) fn with_prefix_hit(mut self, tokens: usize) -> Self { + self.prefix_hit_tokens = tokens; + self + } + pub(crate) fn with_stop_token(mut self, token: u32) -> Self { self.stop_token = Some(token); self @@ -101,7 +113,13 @@ impl FakeExecutor { first_token: 100 + req.request_id.raw() as u32, first_token_logprob: None, prompt_logprobs: None, - cached_tokens: 0, + // A simulated prefix-cache hit is reported only on the request's + // first chunk (start == 0); later chunks carry no cached prefix. + cached_tokens: if start == 0 { + self.prefix_hit_tokens + } else { + 0 + }, completed, prefill_pos, }