Merge pull request 'feat(helexa-bench): percentiles + prefill/decode split in store & report (#86)' (#101) from feat/86-bench-percentiles into main

Bench reported only a single median per cell, hiding tail latency and
unable to record the server-measured prefill/decode split now emitted
on `usage.helexa_timing` (#85).

- scenario: parse `usage.helexa_timing` into ScenarioMetrics
  (prefill_ms, decode_ms, prefill_tokens).
- store: persist the three columns (additive PRAGMA-guarded migration
  via ensure_columns, so pre-#85 DBs backfill as NULL); aggregate now
  emits p50/p95/p99 for TTFT and total (nearest-rank) plus a
  prefill-tok/s median derived from the split.
- report: markdown gains prefill tok/s, TTFT p95, total p95 columns;
  JSON gains p95/p99 + prefill_ms/decode_ms/prefill_tps medians.

Tests: nearest-rank percentile, idempotent backfill migration, and a
report cell asserting percentiles + prefill split.

Part of the Performance observability epic (#83). Stacked on #85.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VrJ4i3pfLRSTM76o3ofnVq

crates/helexa-bench/src/report.rs

@@ -9,22 +9,26 @@ use anyhow::Result;
 pub fn render_markdown(rows: &[ReportRow]) -> String {
     let mut out = String::new();
     out.push_str(
-        "| engine | model | prompt tok | TTFT (s) | decode tok/s | total (s) | build | n |\n",
+        "| engine | model | prompt tok | prefill tok/s | TTFT (s) | TTFT p95 | \
+         decode tok/s | total (s) | total p95 | build | n |\n",
     );
-    out.push_str("|---|---|---:|---:|---:|---:|---|---:|\n");
+    out.push_str("|---|---|---:|---:|---:|---:|---:|---:|---:|---|---:|\n");
     for r in rows {
         let ptok = r
             .prompt_tokens
             .map(|t| t.to_string())
             .unwrap_or_else(|| format!("~{}", r.prompt_size_approx));
         out.push_str(&format!(
-            "| {} | {} | {} | {} | {} | {} | `{}` | {} |\n",
+            "| {} | {} | {} | {} | {} | {} | {} | {} | {} | `{}` | {} |\n",
             r.target_name,
             r.model_id,
             ptok,
+            fmt_opt(r.prefill_tps_median, 1),
             fmt_opt(r.ttft_s_median, 3),
+            fmt_opt(r.ttft_s_p95, 3),
             fmt_opt(r.decode_tps_median, 1),
             fmt_opt(r.total_s_median, 3),
+            fmt_opt(r.total_s_p95, 3),
             r.git_sha,
             r.samples,
         ));
@@ -43,8 +47,15 @@ pub fn render_json(rows: &[ReportRow]) -> Result<String> {
                 "prompt_size_approx": r.prompt_size_approx,
                 "prompt_tokens": r.prompt_tokens,
                 "ttft_s_median": r.ttft_s_median,
+                "ttft_s_p95": r.ttft_s_p95,
+                "ttft_s_p99": r.ttft_s_p99,
                 "decode_tps_median": r.decode_tps_median,
                 "total_s_median": r.total_s_median,
+                "total_s_p95": r.total_s_p95,
+                "total_s_p99": r.total_s_p99,
+                "prefill_ms_median": r.prefill_ms_median,
+                "decode_ms_median": r.decode_ms_median,
+                "prefill_tps_median": r.prefill_tps_median,
                 "git_sha": r.git_sha,
                 "samples": r.samples,
                 "gpu": r.gpu,
@@ -77,14 +88,24 @@ mod tests {
             ttft_s_median: Some(0.123),
             decode_tps_median: Some(45.6),
             total_s_median: Some(1.234),
+            ttft_s_p95: Some(0.222),
+            ttft_s_p99: Some(0.250),
+            total_s_p95: Some(1.5),
+            total_s_p99: Some(1.6),
+            prefill_ms_median: Some(120.0),
+            decode_ms_median: Some(1100.0),
+            prefill_tps_median: Some(1066.7),
             samples: 5,
             gpu: Some("2× RTX 5090".into()),
         }];
         let md = render_markdown(&rows);
         assert!(md.contains("| engine |"));
+        assert!(md.contains("prefill tok/s"));
         assert!(md.contains("beast"));
         assert!(md.contains("`30d50d6`"));
         assert!(md.contains("0.123"));
+        // p95 column rendered.
+        assert!(md.contains("0.222"));
     }
 
     #[test]
@@ -99,6 +120,13 @@ mod tests {
             ttft_s_median: Some(0.1),
             decode_tps_median: None,
             total_s_median: Some(0.5),
+            ttft_s_p95: Some(0.1),
+            ttft_s_p99: Some(0.1),
+            total_s_p95: Some(0.5),
+            total_s_p99: Some(0.5),
+            prefill_ms_median: None,
+            decode_ms_median: None,
+            prefill_tps_median: None,
             samples: 1,
             gpu: None,
         }];

crates/helexa-bench/src/scenario.rs

@@ -62,6 +62,16 @@ pub struct ScenarioMetrics {
     pub prompt_tokens: Option<u64>,
     /// Completion tokens: from `usage` when present, else content-chunk count.
     pub completion_tokens: u64,
+    /// Server-measured prefill duration (ms), from the `usage.helexa_timing`
+    /// extension (#85). `None` when the server didn't emit it (external
+    /// engines, non-instrumented paths). The honest prefill-phase number,
+    /// distinct from client-observed `ttft_s` which also includes request
+    /// setup + first-byte network latency.
+    pub prefill_ms: Option<u64>,
+    /// Server-measured decode duration (ms), from `usage.helexa_timing`.
+    pub decode_ms: Option<u64>,
+    /// Tokens submitted to prefill — the denominator for prefill tok/s.
+    pub prefill_tokens: Option<u64>,
 }
 
 #[async_trait]
@@ -160,6 +170,9 @@ async fn stream_and_measure(
     let mut chunk_count: u64 = 0;
     let mut prompt_tokens: Option<u64> = None;
     let mut completion_tokens: Option<u64> = None;
+    let mut prefill_ms: Option<u64> = None;
+    let mut decode_ms: Option<u64> = None;
+    let mut prefill_tokens: Option<u64> = None;
 
     while let Some(event) = stream.next().await {
         let event = event.context("reading SSE stream")?;
@@ -188,6 +201,11 @@ async fn stream_and_measure(
         if let Some(usage) = chunk.usage {
             prompt_tokens = Some(usage.prompt_tokens);
             completion_tokens = Some(usage.completion_tokens);
+            if let Some(t) = usage.helexa_timing {
+                prefill_ms = Some(t.prefill_ms);
+                decode_ms = Some(t.decode_ms);
+                prefill_tokens = Some(t.prefill_tokens);
+            }
         }
     }
     let end = Instant::now();
@@ -212,6 +230,9 @@ async fn stream_and_measure(
         total_s: (end - start).as_secs_f64(),
         prompt_tokens,
         completion_tokens: tokens,
+        prefill_ms,
+        decode_ms,
+        prefill_tokens,
     })
 }
 

crates/helexa-bench/src/store.rs

@@ -51,6 +51,11 @@ pub struct RunRecord {
     pub decode_tps: Option<f64>,
     pub total_s: Option<f64>,
     pub completion_tokens: Option<u64>,
+    // server-measured prefill/decode split (#85), null on engines/paths
+    // that don't emit `usage.helexa_timing`.
+    pub prefill_ms: Option<u64>,
+    pub decode_ms: Option<u64>,
+    pub prefill_tokens: Option<u64>,
     // outcome
     pub ok: bool,
     pub error: Option<String>,
@@ -123,6 +128,9 @@ impl Store {
                 decode_tps           REAL,
                 total_s              REAL,
                 completion_tokens    INTEGER,
+                prefill_ms           INTEGER,
+                decode_ms            INTEGER,
+                prefill_tokens       INTEGER,
                 ok                   INTEGER NOT NULL,
                 error                TEXT
             );
@@ -133,6 +141,39 @@ impl Store {
             "#,
         )
         .context("initialising sqlite schema")?;
+        // Additive migrations for DBs created before a column existed.
+        // `CREATE TABLE IF NOT EXISTS` above only seeds fresh DBs; existing
+        // ones need the columns backfilled (as NULL) so older rows coexist
+        // with new metrics. There is no migration framework — each entry is
+        // an idempotent "add if missing".
+        Self::ensure_columns(
+            conn,
+            "runs",
+            &[
+                ("prefill_ms", "INTEGER"),
+                ("decode_ms", "INTEGER"),
+                ("prefill_tokens", "INTEGER"),
+            ],
+        )?;
+        Ok(())
+    }
+
+    /// Add any of `columns` that the table is missing (`ALTER TABLE ADD
+    /// COLUMN`). Idempotent: existing columns are read from
+    /// `PRAGMA table_info` and skipped, so this is safe to run on every open.
+    fn ensure_columns(conn: &Connection, table: &str, columns: &[(&str, &str)]) -> Result<()> {
+        let mut existing = std::collections::HashSet::new();
+        let mut stmt = conn.prepare(&format!("PRAGMA table_info({table})"))?;
+        let names = stmt.query_map([], |row| row.get::<_, String>(1))?;
+        for name in names {
+            existing.insert(name?);
+        }
+        for (name, ty) in columns {
+            if !existing.contains(*name) {
+                conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {name} {ty};"))
+                    .with_context(|| format!("adding column {table}.{name}"))?;
+            }
+        }
         Ok(())
     }
 
@@ -166,6 +207,7 @@ impl Store {
                 model_id, harness, capabilities_json, devices_json,
                 scenario_id, prompt_size_approx, prompt_tokens_actual, max_tokens,
                 ttft_s, decode_tps, total_s, completion_tokens,
+                prefill_ms, decode_ms, prefill_tokens,
                 ok, error
             ) VALUES (
                 ?1, ?2, ?3, ?4,
@@ -176,7 +218,8 @@ impl Store {
                 ?20, ?21, ?22, ?23,
                 ?24, ?25, ?26, ?27,
                 ?28, ?29, ?30, ?31,
-                ?32, ?33
+                ?32, ?33, ?34,
+                ?35, ?36
             )",
             params![
                 r.ts,
@@ -210,6 +253,9 @@ impl Store {
                 r.decode_tps,
                 r.total_s,
                 r.completion_tokens,
+                r.prefill_ms,
+                r.decode_ms,
+                r.prefill_tokens,
                 r.ok as i64,
                 r.error,
             ],
@@ -224,7 +270,8 @@ impl Store {
         // successful run, then median that SHA's samples.
         let mut stmt = self.conn.prepare(
             "SELECT target_name, model_id, scenario_id, prompt_size_approx, git_sha,
-                    ttft_s, decode_tps, total_s, prompt_tokens_actual, gpus_json
+                    ttft_s, decode_tps, total_s, prompt_tokens_actual, gpus_json,
+                    prefill_ms, decode_ms, prefill_tokens
              FROM runs
              WHERE ok=1
              ORDER BY target_name, model_id, scenario_id, id",
@@ -241,6 +288,9 @@ impl Store {
                 total_s: row.get(7)?,
                 prompt_tokens_actual: row.get(8)?,
                 gpus_json: row.get(9)?,
+                prefill_ms: row.get(10)?,
+                decode_ms: row.get(11)?,
+                prefill_tokens: row.get(12)?,
             })
         })?;
         let raws: Vec<RawRow> = rows.collect::<rusqlite::Result<_>>()?;
@@ -379,7 +429,7 @@ impl Store {
             "SELECT id, ts, target_name, hostname, git_sha, build_timestamp, package_version,
                     model_id, harness, scenario_id, prompt_size_approx, prompt_tokens_actual,
                     max_tokens, ttft_s, decode_tps, total_s, completion_tokens, ok, error,
-                    gpus_json
+                    gpus_json, prefill_ms, decode_ms, prefill_tokens
              FROM runs",
         );
         let mut conds: Vec<String> = Vec::new();
@@ -435,6 +485,9 @@ impl Store {
                     completion_tokens: r.get(16)?,
                     ok: r.get::<_, i64>(17)? != 0,
                     error: r.get(18)?,
+                    prefill_ms: r.get(20)?,
+                    decode_ms: r.get(21)?,
+                    prefill_tokens: r.get(22)?,
                 })
             })?
             .collect::<rusqlite::Result<_>>()?;
@@ -554,6 +607,9 @@ pub struct RunRow {
     pub decode_tps: Option<f64>,
     pub total_s: Option<f64>,
     pub completion_tokens: Option<u64>,
+    pub prefill_ms: Option<u64>,
+    pub decode_ms: Option<u64>,
+    pub prefill_tokens: Option<u64>,
     pub ok: bool,
     pub error: Option<String>,
 }
@@ -569,6 +625,9 @@ struct RawRow {
     total_s: Option<f64>,
     prompt_tokens_actual: Option<u64>,
     gpus_json: Option<String>,
+    prefill_ms: Option<u64>,
+    decode_ms: Option<u64>,
+    prefill_tokens: Option<u64>,
 }
 
 /// An aggregated cell ready for the report table.
@@ -583,6 +642,19 @@ pub struct ReportRow {
     pub ttft_s_median: Option<f64>,
     pub decode_tps_median: Option<f64>,
     pub total_s_median: Option<f64>,
+    /// Latency tail percentiles — where batch-1 pain actually shows up, and
+    /// invisible behind a bare median. p95/p99 nearest-rank; with few
+    /// samples they collapse toward the max (honest, not interpolated).
+    pub ttft_s_p95: Option<f64>,
+    pub ttft_s_p99: Option<f64>,
+    pub total_s_p95: Option<f64>,
+    pub total_s_p99: Option<f64>,
+    /// Server-measured prefill/decode split (#85). `prefill_tps_median` is
+    /// the true prompt-encoding rate (prefill_tokens / prefill_ms),
+    /// complementing `decode_tps_median` (the generation rate).
+    pub prefill_ms_median: Option<f64>,
+    pub decode_ms_median: Option<f64>,
+    pub prefill_tps_median: Option<f64>,
     pub samples: usize,
     /// Public-facing resource name (the host's GPU(s)), e.g. "2× RTX 5090".
     pub gpu: Option<String>,
@@ -611,6 +683,11 @@ fn aggregate(raws: Vec<RawRow>) -> Vec<ReportRow> {
         let latest_sha = rows.last().map(|r| r.git_sha.clone()).unwrap_or_default();
         let cell: Vec<&RawRow> = rows.iter().filter(|r| r.git_sha == latest_sha).collect();
         let prompt_size_approx = cell.first().map(|r| r.prompt_size_approx).unwrap_or(0);
+        // Per-row prefill tok/s, derived from the server-measured split.
+        let prefill_tps = |r: &&RawRow| match (r.prefill_tokens, r.prefill_ms) {
+            (Some(tok), Some(ms)) if ms > 0 => Some(tok as f64 * 1000.0 / ms as f64),
+            _ => None,
+        };
         out.push(ReportRow {
             target_name,
             model_id,
@@ -621,6 +698,13 @@ fn aggregate(raws: Vec<RawRow>) -> Vec<ReportRow> {
             ttft_s_median: median(cell.iter().filter_map(|r| r.ttft_s)),
             decode_tps_median: median(cell.iter().filter_map(|r| r.decode_tps)),
             total_s_median: median(cell.iter().filter_map(|r| r.total_s)),
+            ttft_s_p95: percentile(cell.iter().filter_map(|r| r.ttft_s), 95.0),
+            ttft_s_p99: percentile(cell.iter().filter_map(|r| r.ttft_s), 99.0),
+            total_s_p95: percentile(cell.iter().filter_map(|r| r.total_s), 95.0),
+            total_s_p99: percentile(cell.iter().filter_map(|r| r.total_s), 99.0),
+            prefill_ms_median: median(cell.iter().filter_map(|r| r.prefill_ms.map(|m| m as f64))),
+            decode_ms_median: median(cell.iter().filter_map(|r| r.decode_ms.map(|m| m as f64))),
+            prefill_tps_median: median(cell.iter().filter_map(prefill_tps)),
             samples: cell.len(),
             gpu: cell
                 .iter()
@@ -680,6 +764,22 @@ fn median(values: impl Iterator<Item = f64>) -> Option<f64> {
     Some((v[lo] + v[hi]) / 2.0)
 }
 
+/// Nearest-rank percentile (`p` in 0..=100). Chosen over interpolation
+/// because bench cells hold only a handful of samples: with n=5, p95/p99
+/// resolve to the max, which honestly says "this is the worst we saw"
+/// rather than inventing a value between samples we never observed.
+fn percentile(values: impl Iterator<Item = f64>, p: f64) -> Option<f64> {
+    let mut v: Vec<f64> = values.collect();
+    if v.is_empty() {
+        return None;
+    }
+    v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
+    // rank = ceil(p/100 * n), clamped to [1, n]; index is rank-1.
+    let rank = (p / 100.0 * v.len() as f64).ceil() as usize;
+    let idx = rank.clamp(1, v.len()) - 1;
+    Some(v[idx])
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -717,6 +817,9 @@ mod tests {
             decode_tps: Some(50.0),
             total_s: Some(1.0),
             completion_tokens: Some(50),
+            prefill_ms: Some(200),
+            decode_ms: Some(1000),
+            prefill_tokens: Some(130),
             ok,
             error: if ok { None } else { Some("boom".into()) },
         }
@@ -755,6 +858,66 @@ mod tests {
         assert!((rows[0].ttft_s_median.unwrap() - 0.3).abs() < 1e-9);
     }
 
+    #[test]
+    fn report_surfaces_percentiles_and_prefill_split() {
+        let s = Store::open_in_memory().unwrap();
+        // Five samples on one cell with spread TTFT so percentiles differ
+        // from the median, plus a server-measured prefill/decode split.
+        for (i, ttft) in [0.10, 0.12, 0.14, 0.16, 0.50].iter().enumerate() {
+            let mut r = rec("beast", "sha", "m", "chat:128", true);
+            r.ttft_s = Some(*ttft);
+            r.total_s = Some(ttft + 1.0);
+            r.prefill_ms = Some(200 + i as u64);
+            r.prefill_tokens = Some(400);
+            s.insert_run(&r).unwrap();
+        }
+        let rows = s.report_rows().unwrap();
+        assert_eq!(rows.len(), 1);
+        let row = &rows[0];
+        assert_eq!(row.samples, 5);
+        // p50 is the middle value; p95/p99 (nearest-rank, n=5) hit the max.
+        assert!((row.ttft_s_median.unwrap() - 0.14).abs() < 1e-9);
+        assert!((row.ttft_s_p95.unwrap() - 0.50).abs() < 1e-9);
+        assert!((row.ttft_s_p99.unwrap() - 0.50).abs() < 1e-9);
+        // prefill tok/s = 400 tok / ~0.2 s ≈ 2000 tok/s.
+        assert!(row.prefill_tps_median.unwrap() > 1900.0);
+        assert!(row.prefill_ms_median.is_some());
+    }
+
+    #[test]
+    fn percentile_nearest_rank() {
+        let vals = || [1.0, 2.0, 3.0, 4.0, 5.0].into_iter();
+        assert_eq!(percentile(vals(), 50.0), Some(3.0));
+        assert_eq!(percentile(vals(), 95.0), Some(5.0));
+        assert_eq!(percentile(vals(), 99.0), Some(5.0));
+        assert_eq!(percentile(std::iter::empty(), 95.0), None);
+    }
+
+    #[test]
+    fn migration_is_idempotent_and_backfills() {
+        // A DB whose `runs` table predates the prefill columns: create the
+        // pre-#85 shape, insert a row, then run ensure_columns twice.
+        let conn = Connection::open_in_memory().unwrap();
+        conn.execute_batch(
+            "CREATE TABLE runs (id INTEGER PRIMARY KEY, ttft_s REAL);
+             INSERT INTO runs (ttft_s) VALUES (0.1);",
+        )
+        .unwrap();
+        for _ in 0..2 {
+            Store::ensure_columns(
+                &conn,
+                "runs",
+                &[("prefill_ms", "INTEGER"), ("decode_ms", "INTEGER")],
+            )
+            .unwrap();
+        }
+        // Columns now exist and the old row reads them back as NULL.
+        let got: Option<i64> = conn
+            .query_row("SELECT prefill_ms FROM runs", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(got, None);
+    }
+
     #[test]
     fn gpu_label_formats() {
         let two = r#"[{"name":"NVIDIA GeForce RTX 5090"},{"name":"NVIDIA GeForce RTX 5090"}]"#;

crates/helexa-bench/src/sweep.rs

@@ -9,7 +9,7 @@
 
 use crate::client::TargetClient;
 use crate::config::{BenchConfig, TargetConfig, TargetKind};
-use crate::scenario::{RunCtx, build_scenarios};
+use crate::scenario::{RunCtx, ScenarioMetrics, build_scenarios};
 use crate::store::{RunRecord, Store};
 use anyhow::Result;
 use cortex_core::build_info::BuildInfo;
@@ -187,18 +187,11 @@ impl Sweeper {
         prompt_size: u32,
         result: Result<&crate::scenario::ScenarioMetrics, &str>,
     ) -> RunRecord {
-        let (ok, error, ttft, decode, total, prompt_tokens, completion) = match result {
-            Ok(m) => (
-                true,
-                None,
-                Some(m.ttft_s),
-                m.decode_tps,
-                Some(m.total_s),
-                m.prompt_tokens,
-                Some(m.completion_tokens),
-            ),
-            Err(e) => (false, Some(e.to_string()), None, None, None, None, None),
+        let (m, error): (Option<&ScenarioMetrics>, Option<String>) = match result {
+            Ok(m) => (Some(m), None),
+            Err(e) => (None, Some(e.to_string())),
         };
+        let ok = m.is_some();
 
         RunRecord {
             ts: chrono::Utc::now().to_rfc3339(),
@@ -230,12 +223,15 @@ impl Sweeper {
                 .unwrap_or_else(|_| "[]".to_string()),
             scenario_id: scenario_id.to_string(),
             prompt_size_approx: prompt_size,
-            prompt_tokens_actual: prompt_tokens,
+            prompt_tokens_actual: m.and_then(|m| m.prompt_tokens),
             max_tokens: self.cfg.scenarios.max_tokens,
-            ttft_s: ttft,
-            decode_tps: decode,
-            total_s: total,
-            completion_tokens: completion,
+            ttft_s: m.map(|m| m.ttft_s),
+            decode_tps: m.and_then(|m| m.decode_tps),
+            total_s: m.map(|m| m.total_s),
+            completion_tokens: m.map(|m| m.completion_tokens),
+            prefill_ms: m.and_then(|m| m.prefill_ms),
+            decode_ms: m.and_then(|m| m.decode_ms),
+            prefill_tokens: m.and_then(|m| m.prefill_tokens),
             ok,
             error,
         }

crates/helexa-bench/tests/api.rs

@@ -46,6 +46,9 @@ fn rec(
         decode_tps: if ok { Some(30.0) } else { None },
         total_s: if ok { Some(2.0) } else { None },
         completion_tokens: if ok { Some(60) } else { None },
+        prefill_ms: if ok { Some(150) } else { None },
+        decode_ms: if ok { Some(1800) } else { None },
+        prefill_tokens: if ok { Some(130) } else { None },
         ok,
         error: if ok { None } else { Some("boom".into()) },
     }