fix(bench): one identity per stream, or fair-share caps the burst at 2

Authenticating bench (#288) fixed the anonymous-yield problem and
introduced a worse one. Fair-share (#54) bounds a single principal to
`max_per_principal` requests in-flight-or-queued; it defaults to 2 and
beast does not override it. Eight streams under one key would have had
six rejected with `PrincipalCap` — so `concurrency:8` would have
measured the fair-share cap instead of the server, and reported it as
backpressure, which reads as a healthy node shedding load rather than an
instrument measuring itself.

Each stream in a burst now presents its own key (`<key_id>-<n>`).

That is also the more faithful model. A concurrency-8 cell is meant to
stand in for eight simultaneous *callers*, not one caller issuing eight
requests, and eight callers is what the router and admission would see in
production. Fair-share is still exercised — every identity is subject to
it — just not self-inflicted by the measurement.

The base identity stays a default header on the client so every scenario
is identified without having to remember; the burst overrides only the
key id per request. Single-request scenarios add no override at all,
since a per-request copy would be a second place for the value to drift
from the client's.

Two tests pin it: a burst yields distinct per-stream keys, and a
single-request scenario adds none.

Worth recording for whoever sets `[bench.principal]`: the account/key
pair needs no provisioning anywhere. `principal_key` in neuron
concatenates the two headers into an opaque fair-share key with no
validation and no allowlist, and bench talks directly to a neuron, so
cortex's entitlements never see it. What the pair *does* need is to be
distinct from any real caller's, so bench holds its own allocation
rather than eating someone else's.

Refs #288, #54, #262

asset/helexa-bench/bob.toml

@@ -15,6 +15,20 @@ iteration_pause_secs = 2
 request_timeout_secs = 600
 db_path = "/var/lib/helexa-bench/bench.sqlite"
 
+# Identity bench presents to neurons (#288). Without it bench is
+# anonymous, and since #262 anonymous callers are served only from
+# leftover capacity — capped below max_in_flight and parked at the class
+# gate. That moved beast's concurrency:8 from 168.9 tok/s / 0.53s
+# ttft_p95 to 104.6 / 11.77s the day #262 landed and held there, so every
+# number since measured the yield policy, not serving capacity.
+#
+# A dedicated pair, not a borrowed one: bench then holds its own
+# fair-share allocation (#54) and cannot starve interactive traffic —
+# the failure #262 existed to fix.
+[bench.principal]
+account_id = "helexa-bench"
+key_id = "fleet-benchmark"
+
 [scenarios]
 prompt_sizes = [128, 4096]
 max_tokens = 256
@@ -22,6 +36,10 @@ max_tokens = 256
 # carries p95-under-concurrency data before the F3 A/B gate (#94) needs
 # it for comparison. Levels mirror the real a0/hermes/opencode fan-out.
 concurrency_levels = [2, 4, 8]
+# Non-streaming bursts (#285/#288). Kept to the top level: it is where
+# serialization shows most starkly and each level costs a full burst per
+# sample per build.
+concurrency_nonstreaming_levels = [8]
 concurrency_prompt_tokens = 512
 
 # Capability probes (#91) — the reasoning/planning axis the speed

bench/src/pages/Trends.tsx

@@ -40,6 +40,80 @@ function Picker({
   );
 }
 
+type SeriesDef = {
+  key: string;
+  name: string;
+  stroke: string;
+  dashed?: boolean;
+};
+
+/** One titled chart over the shared build timeline.
+ *
+ * Every panel draws the same x-axis and the same regime divider, so a
+ * reader can line a change up across metrics — which is the whole point
+ * of having more than two of them. */
+function MetricChart({
+  title,
+  hint,
+  data,
+  lines,
+  divider,
+  unit,
+}: {
+  title: string;
+  hint?: string;
+  data: Record<string, unknown>[];
+  lines: SeriesDef[];
+  divider?: string;
+  unit?: string;
+}) {
+  const hasAny = data.some((d) => lines.some((l) => d[l.key] != null));
+  if (!hasAny) return null;
+  return (
+    <>
+      <h5 className="mt-4">
+        {title}
+        {unit ? <span className="text-muted fw-normal"> ({unit})</span> : null}
+      </h5>
+      {hint && <p className="text-muted small mb-2">{hint}</p>}
+      <ResponsiveContainer width="100%" height={260}>
+        <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
+          <CartesianGrid strokeDasharray="3 3" />
+          <XAxis dataKey="label" />
+          <YAxis />
+          <Tooltip />
+          <Legend />
+          {divider && (
+            <ReferenceLine
+              x={divider}
+              stroke="#bbb"
+              strokeDasharray="3 3"
+              label={{
+                value: "bench.py → helexa-bench",
+                position: "top",
+                fill: "#999",
+                fontSize: 11,
+              }}
+            />
+          )}
+          {lines.map((l) => (
+            <Line
+              key={l.key}
+              type="monotone"
+              dataKey={l.key}
+              name={l.name}
+              stroke={l.stroke}
+              strokeDasharray={l.dashed ? "5 5" : undefined}
+              connectNulls
+              dot={false}
+            />
+          ))}
+        </LineChart>
+      </ResponsiveContainer>
+    </>
+  );
+}
+
 export default function Trends() {
   const [dims, setDims] = useState<Dimensions | null>(null);
   const [model, setModel] = useState("");
@@ -68,10 +142,7 @@ export default function Trends() {
   // Prepend the pre-helexa-bench baseline (dashed, separate keys) so it
   // anchors the timeline without being merged into the live line. Different
   // measurement regime — see baseline.ts / doc/benchmarks.md.
-  const base = useMemo(
-    () => baselineFor(model, scenario),
-    [model, scenario],
-  );
+  const base = useMemo(() => baselineFor(model, scenario), [model, scenario]);
   const data = useMemo(
     () => [
       ...base.map((p) => ({
@@ -85,6 +156,14 @@ export default function Trends() {
         ttft: p.ttft_s_median,
         decode: p.decode_tps_median,
         total: p.total_s_median,
+        ttftP95: p.ttft_p95_s_median,
+        queueWait: p.queue_wait_ms_median,
+        rejected: p.rejected_median,
+        prefillTps: p.prefill_tps_median,
+        reasoning: p.reasoning_tokens_median,
+        cached: p.cached_tokens_median,
+        completion: p.completion_tokens_median,
+        tpot: p.tpot_p95_ms_median,
       })),
     ],
     [series, base],
@@ -94,6 +173,19 @@ export default function Trends() {
   // first live build, with baseline points to its left).
   const firstLive = series[0]?.git_sha;
   const showDivider = base.length > 0 && series.length > 0;
+  const divider = showDivider ? firstLive : undefined;
+
+  // Anonymous and identified samples are not comparable once #262 is in
+  // the build: an anonymous caller is capped below max_in_flight and
+  // parked at the class gate, so it characterises the yield policy
+  // rather than serving capacity. Say so rather than letting someone
+  // read a step change as an engine regression (#288).
+  const identities = useMemo(
+    () => new Set(series.map((p) => p.principal ?? "anonymous")),
+    [series],
+  );
+  const mixedIdentity = identities.size > 1;
+  const anyAnonymous = identities.has("anonymous");
 
   if (err) return <Alert variant="danger">{err}</Alert>;
   if (!dims) return <Spinner animation="border" />;
@@ -122,6 +214,25 @@ export default function Trends() {
         </p>
       )}
 
+      {mixedIdentity && (
+        <Alert variant="warning" className="py-2">
+          <strong>Mixed measurement identity.</strong> Some builds were
+          sampled anonymously and some under a principal. Since{" "}
+          <code>#262</code> an anonymous caller is capped below{" "}
+          <code>max_in_flight</code> and yields to identified traffic, so
+          those points measure the admission policy rather than serving
+          capacity. A step change across that boundary is the instrument,
+          not the engine — see <code>#288</code>.
+        </Alert>
+      )}
+      {!mixedIdentity && anyAnonymous && (
+        <Alert variant="secondary" className="py-2 small">
+          Sampled anonymously. Since <code>#262</code> anonymous callers are
+          served from leftover capacity, so these numbers understate what an
+          authenticated caller gets (<code>#288</code>).
+        </Alert>
+      )}
+
       {data.length === 0 ? (
         <Alert variant="info">No data for this selection yet.</Alert>
       ) : (
@@ -133,87 +244,95 @@ export default function Trends() {
               see <code>doc/benchmarks.md</code>.
             </p>
           )}
-          <h5 className="mt-3">decode tok/s (higher is better)</h5>
-          <ResponsiveContainer width="100%" height={280}>
-            <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
-              <CartesianGrid strokeDasharray="3 3" />
-              <XAxis dataKey="label" />
-              <YAxis />
-              <Tooltip />
-              <Legend />
-              {showDivider && firstLive && (
-                <ReferenceLine
-                  x={firstLive}
-                  stroke="#bbb"
-                  strokeDasharray="3 3"
-                  label={{
-                    value: "bench.py → helexa-bench",
-                    position: "top",
-                    fill: "#999",
-                    fontSize: 11,
-                  }}
-                />
-              )}
-              <Line
-                type="monotone"
-                dataKey="decode"
-                name="decode tok/s"
-                stroke="#0d6efd"
-                connectNulls
-              />
-              {base.length > 0 && (
-                <Line
-                  type="monotone"
-                  dataKey="baseDecode"
-                  name="baseline (bench.py · gateway)"
-                  stroke="#888"
-                  strokeDasharray="5 5"
-                  connectNulls
-                />
-              )}
-            </LineChart>
-          </ResponsiveContainer>
 
-          <h5 className="mt-4">TTFT seconds (lower is better)</h5>
-          <ResponsiveContainer width="100%" height={280}>
-            <LineChart data={data} margin={{ top: 8, right: 24, bottom: 8, left: 0 }}>
-              <CartesianGrid strokeDasharray="3 3" />
-              <XAxis dataKey="label" />
-              <YAxis />
-              <Tooltip />
-              <Legend />
-              {showDivider && firstLive && (
-                <ReferenceLine
-                  x={firstLive}
-                  stroke="#bbb"
-                  strokeDasharray="3 3"
-                  label={{
-                    value: "bench.py → helexa-bench",
-                    position: "top",
-                    fill: "#999",
-                    fontSize: 11,
-                  }}
-                />
-              )}
-              <Line
-                type="monotone"
-                dataKey="ttft"
-                name="TTFT (s)"
-                stroke="#dc3545"
-                connectNulls
-              />
-              {base.length > 0 && (
-                <Line
-                  type="monotone"
-                  dataKey="baseTtft"
-                  name="baseline (bench.py · gateway)"
-                  stroke="#888"
-                  strokeDasharray="5 5"
-                  connectNulls
-                />
-              )}
-            </LineChart>
-          </ResponsiveContainer>
+          <MetricChart
+            title="decode tok/s"
+            unit="higher is better"
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "decode", name: "decode tok/s", stroke: "#0d6efd" },
+              ...(base.length > 0
+                ? [
+                    {
+                      key: "baseDecode",
+                      name: "baseline (bench.py · gateway)",
+                      stroke: "#888",
+                      dashed: true,
+                    },
+                  ]
+                : []),
+            ]}
+          />
+
+          <MetricChart
+            title="prefill tok/s"
+            unit="higher is better"
+            hint="The other half of serving speed, derived from prefill_tokens / prefill_ms. A prefix-cache hit shortens prefill_ms while the token count stays whole, so a high rate here is itself the cache-hit signal."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "prefillTps", name: "prefill tok/s", stroke: "#20c997" },
+            ]}
+          />
+
+          <MetricChart
+            title="TTFT"
+            unit="seconds, lower is better"
+            hint="Median and p95 together on purpose. Under concurrency the median is dominated by whichever streams were admitted immediately; the p95 is the one that moves when a caller is made to wait. Charting only the median is how a 0.53 s → 11.77 s tail went unnoticed for a week."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "ttft", name: "TTFT median (s)", stroke: "#dc3545" },
+              { key: "ttftP95", name: "TTFT p95 (s)", stroke: "#fd7e14" },
+              ...(base.length > 0
+                ? [
+                    {
+                      key: "baseTtft",
+                      name: "baseline (bench.py · gateway)",
+                      stroke: "#888",
+                      dashed: true,
+                    },
+                  ]
+                : []),
+            ]}
+          />
+
+          <MetricChart
+            title="inter-token gap p95"
+            unit="ms, lower is better"
+            hint="Stream smoothness. decode tok/s is a mean over the whole window, so a stream that stalls and then catches up is indistinguishable from one that never stalled — this is the number a user feels."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "tpot", name: "inter-token p95 (ms)", stroke: "#6f42c1" },
+            ]}
+          />
+
+          <MetricChart
+            title="admission"
+            unit="queue wait ms · requests shed"
+            hint="Separates “the server is slow” from “you were queued behind someone”. Queue wait is TTFT minus server-measured prefill; rejected counts honest backpressure rather than silent failures."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "queueWait", name: "queue wait (ms)", stroke: "#d63384" },
+              { key: "rejected", name: "rejected (count)", stroke: "#adb5bd" },
+            ]}
+          />
+
+          <MetricChart
+            title="tokens per sample"
+            unit="counts"
+            hint="Cost, not speed. Reasoning tokens are the dominant driver on a reasoning model and move independently of every rate above — a template or sampling change can double what the model thinks before answering while the speed charts stay flat. Cached tokens are why prefill timing varies between otherwise identical samples."
+            data={data}
+            divider={divider}
+            lines={[
+              { key: "completion", name: "completion tokens", stroke: "#0dcaf0" },
+              { key: "reasoning", name: "reasoning tokens", stroke: "#ffc107" },
+              { key: "cached", name: "cached prompt tokens", stroke: "#198754" },
+            ]}
+          />
         </>
       )}
     </>

bench/src/types.ts

@@ -41,6 +41,17 @@ export interface SeriesPoint {
   ttft_s_median: number | null;
   decode_tps_median: number | null;
   total_s_median: number | null;
+  ttft_p95_s_median: number | null;
+  queue_wait_ms_median: number | null;
+  rejected_median: number | null;
+  prefill_tps_median: number | null;
+  reasoning_tokens_median: number | null;
+  cached_tokens_median: number | null;
+  completion_tokens_median: number | null;
+  tpot_p95_ms_median: number | null;
+  /** Identity the samples were taken under, or null if anonymous.
+   *  Anonymous and identified points are not comparable (#288). */
+  principal: string | null;
   samples: number;
 }
 

crates/helexa-bench/src/client.rs

@@ -6,6 +6,7 @@ use crate::config::{TargetConfig, TargetKind};
 use anyhow::{Context, Result, anyhow};
 use cortex_core::build_info::BuildInfo;
 use cortex_core::discovery::{DiscoveryResponse, HealthResponse};
+use cortex_core::entitlements::{HEADER_ACCOUNT_ID, HEADER_KEY_ID};
 use cortex_core::harness::{ModelInfo, ModelSpec};
 use cortex_core::openai::ModelsResponse;
 use std::time::Duration;
@@ -19,10 +20,49 @@ pub struct TargetClient {
 
 impl TargetClient {
     pub fn new(request_timeout: Duration) -> Result<Self> {
-        let http = reqwest::Client::builder()
-            .timeout(request_timeout)
-            .build()
-            .context("building HTTP client")?;
+        Self::with_principal(request_timeout, None)
+    }
+
+    /// Build a client that stamps `principal`'s identity headers on every
+    /// request (#288).
+    ///
+    /// Set on the client rather than at each call site so no scenario can
+    /// be added later that silently measures as anonymous — which is the
+    /// failure this fixes, and it was invisible for a week.
+    ///
+    /// `None` keeps the historical anonymous behaviour, which is correct
+    /// for `openai` targets (the headers mean nothing to a foreign
+    /// engine) and for a fresh install with no principal configured.
+    pub fn with_principal(
+        request_timeout: Duration,
+        principal: Option<&crate::config::PrincipalSettings>,
+    ) -> Result<Self> {
+        let mut builder = reqwest::Client::builder().timeout(request_timeout);
+        if let Some(p) = principal {
+            let mut headers = reqwest::header::HeaderMap::new();
+            headers.insert(
+                HEADER_ACCOUNT_ID,
+                reqwest::header::HeaderValue::from_str(&p.account_id)
+                    .context("principal.account_id is not a valid header value")?,
+            );
+            headers.insert(
+                HEADER_KEY_ID,
+                reqwest::header::HeaderValue::from_str(&p.key_id)
+                    .context("principal.key_id is not a valid header value")?,
+            );
+            tracing::info!(
+                account_id = %p.account_id,
+                key_id = %p.key_id,
+                "bench authenticates: measurements reflect identified-caller capacity (#288)"
+            );
+            builder = builder.default_headers(headers);
+        } else {
+            tracing::warn!(
+                "bench has no [bench.principal]: requests are anonymous, so since #262 \
+                 they measure the anonymous-yield policy rather than serving capacity (#288)"
+            );
+        }
+        let http = builder.build().context("building HTTP client")?;
         Ok(TargetClient { http })
     }
 

crates/helexa-bench/src/config.rs

@@ -67,6 +67,44 @@ pub struct BenchSettings {
     /// SQLite system-of-record path.
     #[serde(default = "default_db_path")]
     pub db_path: String,
+    /// Identity bench presents to a neuron (#288).
+    ///
+    /// Without this bench is *anonymous*, and since #262 anonymous
+    /// callers are served only from capacity left over once identified
+    /// traffic is satisfied — they are capped below `max_in_flight` and
+    /// park at the class gate ahead of every other gate. Measured on
+    /// beast, that moved `concurrency:8` from 168.9 tok/s / 0.53 s
+    /// ttft_p95 to 104.6 / 11.77 s the day #262 landed, and held there.
+    ///
+    /// So an unauthenticated bench does not measure serving capacity; it
+    /// measures the yield policy. Real traffic arrives through cortex
+    /// with a principal and is subject to none of it.
+    ///
+    /// Deliberately not an admission *exemption*: bench should be
+    /// subject to the same rules as any caller, or it measures something
+    /// no user experiences — the same class of error in the other
+    /// direction. Give it a principal and let it queue like everyone
+    /// else.
+    #[serde(default)]
+    pub principal: Option<PrincipalSettings>,
+}
+
+/// The account/key pair bench stamps on inference requests (#288).
+///
+/// These are the headers cortex asserts after a bearer resolves (#49).
+/// Bench talks to neurons **directly** over the WireGuard mesh, so it
+/// stamps them itself — the same trust model the link already relies on,
+/// since a neuron accepts them from cortex on exactly that basis.
+///
+/// Give bench its *own* pair rather than borrowing a real caller's: it
+/// then holds its own fair-share allocation (#54) and cannot starve
+/// interactive traffic, which is the failure #262 was fixing.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct PrincipalSettings {
+    /// Stamped as `x-helexa-account-id`.
+    pub account_id: String,
+    /// Stamped as `x-helexa-key-id`.
+    pub key_id: String,
 }
 
 impl Default for BenchSettings {
@@ -77,6 +115,7 @@ impl Default for BenchSettings {
             iteration_pause_secs: default_iter_pause(),
             request_timeout_secs: default_timeout(),
             db_path: default_db_path(),
+            principal: None,
         }
     }
 }
@@ -110,6 +149,18 @@ pub struct ScenarioConfig {
     /// deliberately, e.g. `concurrency_levels = [2, 4, 8]`.
     #[serde(default)]
     pub concurrency_levels: Vec<u32>,
+    /// Concurrency levels to run **non-streaming** (#285). Separate from
+    /// `concurrency_levels` because the two shapes are not
+    /// interchangeable: streaming multiplexes through the batch engine,
+    /// non-streaming did not until #285 and serialized completely — a
+    /// measured 1.00x aggregate throughput at every level against 3.98x
+    /// streamed.
+    ///
+    /// Defaults to the top level only. The signal is starkest there and
+    /// each extra level costs a full burst per sample per build; an
+    /// operator who wants the whole curve can list it.
+    #[serde(default = "default_concurrency_nonstreaming_levels")]
+    pub concurrency_nonstreaming_levels: Vec<u32>,
     /// Square image sizes (px per side) — one `image:<px>` scenario per
     /// entry (#203), run only against models advertising the `image`
     /// capability. Defaults to `[1024]`.
@@ -148,6 +199,7 @@ impl Default for ScenarioConfig {
             prompt_sizes: default_prompt_sizes(),
             max_tokens: default_max_tokens(),
             concurrency_levels: Vec::new(),
+            concurrency_nonstreaming_levels: Vec::new(),
             concurrency_prompt_tokens: default_concurrency_prompt_tokens(),
             image_sizes: default_image_sizes(),
             capability_probes: Vec::new(),
@@ -215,6 +267,10 @@ fn default_timeout() -> u64 {
 fn default_db_path() -> String {
     "/var/lib/helexa-bench/bench.sqlite".to_string()
 }
+fn default_concurrency_nonstreaming_levels() -> Vec<u32> {
+    vec![8]
+}
+
 fn default_api_enabled() -> bool {
     true
 }
@@ -243,6 +299,44 @@ fn default_capability_max_tokens() -> u64 {
 // figment's, not ours, so suppress the lint here.
 #[allow(clippy::result_large_err)]
 mod tests {
+
+    /// #288: a config with no `[bench.principal]` must still parse — the
+    /// anonymous behaviour is the historical one and correct for
+    /// `openai` targets — but a configured one must round-trip, because
+    /// a silently-dropped principal reverts the measurement to the
+    /// anonymous-yield path without changing a visible number.
+    #[test]
+    fn principal_is_optional_and_round_trips() {
+        let parse = |t: &str| -> BenchConfig {
+            figment::Figment::new()
+                .merge(figment::providers::Toml::string(t))
+                .extract()
+                .expect("parse")
+        };
+        let none = parse("[bench]\nsamples_per_version = 3\n");
+        assert!(none.bench.principal.is_none());
+
+        let some = parse(
+            "[bench]\nsamples_per_version = 3\n\n\
+             [bench.principal]\naccount_id = \"acct\"\nkey_id = \"k1\"\n",
+        );
+        let p = some.bench.principal.expect("principal");
+        assert_eq!((p.account_id.as_str(), p.key_id.as_str()), ("acct", "k1"));
+    }
+
+    /// The non-streaming burst defaults on (#285). If it defaulted off,
+    /// the path that serialized completely would stay unmeasured on every
+    /// fleet that had not opted in — which is how it went unnoticed.
+    #[test]
+    fn nonstreaming_concurrency_defaults_to_the_top_level() {
+        let cfg: BenchConfig = figment::Figment::new()
+            .merge(figment::providers::Toml::string(
+                "[scenarios]\nprompt_sizes = [128]\n",
+            ))
+            .extract()
+            .expect("parse");
+        assert_eq!(cfg.scenarios.concurrency_nonstreaming_levels, vec![8]);
+    }
     use super::*;
     use figment::Jail;
 

crates/helexa-bench/src/scenario.rs

@@ -9,6 +9,7 @@
 use crate::config::ScenarioConfig;
 use anyhow::{Context, Result, anyhow};
 use async_trait::async_trait;
+use cortex_core::entitlements::HEADER_KEY_ID;
 use cortex_core::harness::ModelInfo;
 use cortex_core::openai::ChatCompletionChunk;
 use eventsource_stream::Eventsource;
@@ -46,6 +47,17 @@ pub struct RunCtx<'a> {
     pub model_id: String,
     pub max_tokens: u64,
     pub timeout: Duration,
+    /// Base `x-helexa-key-id` bench presents (#288), when configured.
+    ///
+    /// The client already sends this as a default header, so every
+    /// scenario is identified without having to remember. A concurrency
+    /// burst overrides it per stream — see
+    /// [`ConcurrencyScenario`] — because fair-share (#54) bounds one
+    /// principal to `max_per_principal` in-flight-or-queued requests,
+    /// which defaults to 2. Eight streams under one identity would be
+    /// capped at two and the rest rejected, measuring the fair-share cap
+    /// instead of the server.
+    pub principal_key_id: Option<String>,
 }
 
 /// Operator-felt metrics for a single measured request.
@@ -72,6 +84,34 @@ pub struct ScenarioMetrics {
     pub decode_ms: Option<u64>,
     /// Tokens submitted to prefill — the denominator for prefill tok/s.
     pub prefill_tokens: Option<u64>,
+    /// Tokens spent inside the reasoning span, from
+    /// `usage.completion_tokens_details.reasoning_tokens` (#223). A
+    /// sub-count of `completion_tokens`.
+    ///
+    /// Worth a trend of its own: on a reasoning model this is the
+    /// dominant cost driver and it moves independently of speed. A
+    /// template change, a sampling change (#283) or a reasoning-budget
+    /// change can double what the model thinks before answering while
+    /// every tok/s number stays flat — the bill doubles and no chart
+    /// moves.
+    pub reasoning_tokens: Option<u64>,
+    /// p95 inter-token arrival gap in milliseconds — the tail of the
+    /// stream's smoothness, client-observed.
+    ///
+    /// `decode_tps` is a mean over the whole decode window, so a stream
+    /// that stalls for a second and then catches up is indistinguishable
+    /// from one that never stalled. This is the number a user actually
+    /// feels, and the one a batching stall or a mid-stream rebatch shows
+    /// up in. `None` for non-streaming, which has no inter-token gaps by
+    /// construction, and for streams too short to have a tail.
+    pub tpot_p95_ms: Option<f64>,
+    /// Prompt tokens served from neuron's prefix KV cache (#269), from
+    /// `usage.prompt_tokens_details.cached_tokens`.
+    ///
+    /// The reason prefill timing varies between otherwise identical
+    /// samples. #269 exists because the saving was invisible and every
+    /// client reported a 0% hit rate; bench then threw the number away.
+    pub cached_tokens: Option<u64>,
     // ── Concurrency / agentic-load fields (#89) ──────────────────────────
     // Set only by the concurrency scenario, which fans out N simultaneous
     // streams to characterize the real a0/hermes/opencode workload that
@@ -203,6 +243,9 @@ impl Scenario for ImageLatencyScenario {
                 .as_u64()
                 .map(|d| d + timing["decode_ms"].as_u64().unwrap_or(0)),
             prefill_tokens: None,
+            reasoning_tokens: None,
+            cached_tokens: None,
+            tpot_p95_ms: None,
             concurrency: None,
             ttft_p95_s: None,
             queue_wait_ms_median: None,
@@ -233,6 +276,18 @@ pub fn build_scenarios(cfg: &ScenarioConfig) -> Vec<Box<dyn Scenario>> {
             id: format!("concurrency:{n}"),
             concurrency: n,
             approx_prompt_tokens: cfg.concurrency_prompt_tokens,
+            streaming: true,
+        }) as Box<dyn Scenario>);
+    }
+    // Non-streaming bursts (#285). A separate cell id so the two shapes
+    // never average together — they measured 1.00x against 3.98x before
+    // #285, and a combined number would have hidden both.
+    for &n in &cfg.concurrency_nonstreaming_levels {
+        scenarios.push(Box::new(ConcurrencyScenario {
+            id: format!("concurrency:{n}:nostream"),
+            concurrency: n,
+            approx_prompt_tokens: cfg.concurrency_prompt_tokens,
+            streaming: false,
         }) as Box<dyn Scenario>);
     }
     for &side in &cfg.image_sizes {
@@ -257,7 +312,7 @@ pub fn build_scenarios(cfg: &ScenarioConfig) -> Vec<Box<dyn Scenario>> {
 pub async fn cold_probe(ctx: &RunCtx<'_>) -> Result<ScenarioMetrics> {
     let prompt = build_prompt(128);
     let payload = chat_payload(ctx, &prompt);
-    tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload))
+    tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload, None))
         .await
         .map_err(|_| anyhow!("cold probe timed out after {:?}", ctx.timeout))?
 }
@@ -295,7 +350,7 @@ impl Scenario for ChatLatencyScenario {
     async fn run(&self, ctx: &RunCtx) -> Result<ScenarioMetrics> {
         let prompt = build_prompt(self.approx_prompt_tokens);
         let payload = chat_payload(ctx, &prompt);
-        let fut = stream_and_measure(ctx, &payload);
+        let fut = stream_and_measure(ctx, &payload, None);
         tokio::time::timeout(ctx.timeout, fut)
             .await
             .map_err(|_| anyhow!("request timed out after {:?}", ctx.timeout))?
@@ -313,6 +368,16 @@ pub struct ConcurrencyScenario {
     id: String,
     concurrency: u32,
     approx_prompt_tokens: u32,
+    /// Whether the burst streams (#285).
+    ///
+    /// Both shapes matter and they are not interchangeable. Streaming
+    /// multiplexes through the batch engine; non-streaming did not until
+    /// #285, and serialized completely — 1.00x aggregate throughput at
+    /// every concurrency level against 3.98x streamed. A bench that only
+    /// streams cannot see that, which is why it went unmeasured: the
+    /// neuron reports `in_flight: 8`, admission accepts, `/health` looks
+    /// healthy, and throughput is single-stream.
+    streaming: bool,
 }
 
 #[async_trait]
@@ -327,13 +392,26 @@ impl Scenario for ConcurrencyScenario {
 
     async fn run(&self, ctx: &RunCtx) -> Result<ScenarioMetrics> {
         let prompt = build_prompt(self.approx_prompt_tokens);
-        let payload = chat_payload(ctx, &prompt);
+        let payload = if self.streaming {
+            chat_payload(ctx, &prompt)
+        } else {
+            nonstreaming_payload(ctx, &prompt)
+        };
 
-        // Fire all streams at once; each is independently timed and capped by
-        // the per-request timeout so one hung stream can't stall the burst.
+        // Fire all requests at once; each is independently timed and capped
+        // by the per-request timeout so one hung request can't stall the
+        // burst.
         let burst_start = Instant::now();
-        let futs = (0..self.concurrency).map(|_| async {
-            tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, &payload)).await
+        let streaming = self.streaming;
+        let payload = &payload;
+        let futs = (0..self.concurrency).map(move |i| async move {
+            // Each stream presents its own identity — see
+            // `with_stream_identity`.
+            if streaming {
+                tokio::time::timeout(ctx.timeout, stream_and_measure(ctx, payload, Some(i))).await
+            } else {
+                tokio::time::timeout(ctx.timeout, request_and_measure(ctx, payload, Some(i))).await
+            }
         });
         let results = futures::future::join_all(futs).await;
         let burst_window = burst_start.elapsed().as_secs_f64();
@@ -351,7 +429,7 @@ impl Scenario for ConcurrencyScenario {
         }
         if streams.is_empty() {
             return Err(anyhow!(
-                "all {} concurrent streams failed ({rejected} shed by admission)",
+                "all {} concurrent requests failed ({rejected} shed by admission)",
                 self.concurrency
             ));
         }
@@ -383,6 +461,18 @@ impl Scenario for ConcurrencyScenario {
             prefill_ms: None,
             decode_ms: None,
             prefill_tokens: None,
+            // Summed across the burst, like `completion_tokens`: the cell
+            // describes the whole burst, so a per-stream figure here
+            // would not compose with it.
+            reasoning_tokens: sum_opt(&streams, |m| m.reasoning_tokens),
+            cached_tokens: sum_opt(&streams, |m| m.cached_tokens),
+            // The worst stream in the burst, not a median of medians:
+            // under load the question is how bad it got for somebody,
+            // and averaging tails is how a stall stays invisible.
+            tpot_p95_ms: streams
+                .iter()
+                .filter_map(|m| m.tpot_p95_ms)
+                .max_by(|a, b| a.total_cmp(b)),
             concurrency: Some(self.concurrency),
             ttft_p95_s: percentile(&ttfts, 95.0),
             queue_wait_ms_median: median(&queue_waits),
@@ -393,6 +483,146 @@ impl Scenario for ConcurrencyScenario {
     }
 }
 
+/// Sum an optional per-stream count across a burst, `None` when no
+/// stream reported it — so "nobody measured it" stays distinguishable
+/// from "measured as zero", the same distinction #269 exists to
+/// preserve.
+fn sum_opt(
+    streams: &[ScenarioMetrics],
+    f: impl Fn(&ScenarioMetrics) -> Option<u64>,
+) -> Option<u64> {
+    let vals: Vec<u64> = streams.iter().filter_map(f).collect();
+    (!vals.is_empty()).then(|| vals.iter().sum())
+}
+
+/// Per-stream identity for a concurrency burst (#288/#54).
+///
+/// Fair-share bounds *one principal* to `max_per_principal`
+/// in-flight-or-queued requests, defaulting to 2. A burst of 8 under a
+/// single identity would therefore have 6 rejected with `PrincipalCap`,
+/// and the scenario would measure the fair-share cap rather than the
+/// server.
+///
+/// Giving each stream its own key is also the more faithful model: a
+/// concurrency-8 burst is meant to stand in for eight simultaneous
+/// *callers*, not one caller issuing eight requests. Fair-share is still
+/// exercised — each identity is subject to it — just not self-inflicted.
+fn with_stream_identity(
+    rb: reqwest::RequestBuilder,
+    ctx: &RunCtx<'_>,
+    stream: Option<u32>,
+) -> reqwest::RequestBuilder {
+    match (ctx.principal_key_id.as_deref(), stream) {
+        (Some(base), Some(i)) => rb.header(HEADER_KEY_ID, format!("{base}-{i}")),
+        // No principal configured, or a single-request scenario: the
+        // client's default headers already carry the base identity.
+        _ => rb,
+    }
+}
+
+/// The non-streaming counterpart to [`chat_payload`] (#285).
+///
+/// `stream_options` is deliberately absent: usage is part of the body on
+/// this shape, and sending the streaming-only option to a strict server
+/// is a needless compatibility risk.
+fn nonstreaming_payload(ctx: &RunCtx, prompt: &str) -> serde_json::Value {
+    json!({
+        "model": ctx.model_id,
+        "messages": [{"role": "user", "content": prompt}],
+        "max_tokens": ctx.max_tokens,
+        "temperature": 0,
+        "stream": false,
+    })
+}
+
+/// Time one non-streaming chat completion (#285).
+///
+/// There is no first-chunk moment on this shape — the whole body arrives
+/// at once — so `ttft_s` is the full request wall-clock rather than a
+/// separate measurement. That is the honest reading: for a non-streaming
+/// caller, time-to-first-token *is* time-to-everything, and it is exactly
+/// what makes the serialization this scenario exists to detect so
+/// expensive.
+///
+/// `decode_tps` is per-request; the burst aggregate is computed by the
+/// caller over the whole window, the same way the streaming path does it.
+async fn request_and_measure(
+    ctx: &RunCtx<'_>,
+    payload: &serde_json::Value,
+    stream: Option<u32>,
+) -> Result<ScenarioMetrics> {
+    let start = Instant::now();
+    let resp = with_stream_identity(ctx.client.post(&ctx.chat_url), ctx, stream)
+        .json(payload)
+        .send()
+        .await
+        .context("sending non-streaming chat request")?;
+    if !resp.status().is_success() {
+        let status = resp.status();
+        let body = resp.text().await.unwrap_or_default();
+        return Err(anyhow!("upstream returned {status}: {}", body.trim()));
+    }
+    let v: serde_json::Value = resp
+        .json()
+        .await
+        .context("non-streaming chat response was not JSON")?;
+    let total_s = start.elapsed().as_secs_f64();
+
+    let usage = v.get("usage");
+    let completion_tokens = usage
+        .and_then(|u| u.get("completion_tokens"))
+        .and_then(|t| t.as_u64())
+        .unwrap_or(0);
+    let prompt_tokens = usage
+        .and_then(|u| u.get("prompt_tokens"))
+        .and_then(|t| t.as_u64());
+    let reasoning_tokens = usage
+        .and_then(|u| u.get("completion_tokens_details"))
+        .and_then(|d| d.get("reasoning_tokens"))
+        .and_then(|t| t.as_u64());
+    let cached_tokens = usage
+        .and_then(|u| u.get("prompt_tokens_details"))
+        .and_then(|d| d.get("cached_tokens"))
+        .and_then(|t| t.as_u64());
+    let timing = usage.and_then(|u| u.get("helexa_timing"));
+    let field = |name: &str| timing.and_then(|t| t.get(name)).and_then(|x| x.as_u64());
+    let (prefill_ms, decode_ms, prefill_tokens) = (
+        field("prefill_ms"),
+        field("decode_ms"),
+        field("prefill_tokens"),
+    );
+
+    // Prefer the server's own decode window when it reports one (#85);
+    // it excludes prefill, so it is comparable with the streaming path's
+    // decode-window rate rather than being diluted by it.
+    let decode_tps = match decode_ms {
+        Some(ms) if ms > 200 => Some(completion_tokens as f64 / (ms as f64 / 1000.0)),
+        _ if total_s > 0.2 => Some(completion_tokens as f64 / total_s),
+        _ => None,
+    };
+
+    Ok(ScenarioMetrics {
+        ttft_s: total_s,
+        decode_tps,
+        total_s,
+        prompt_tokens,
+        completion_tokens,
+        prefill_ms,
+        decode_ms,
+        prefill_tokens,
+        reasoning_tokens,
+        cached_tokens,
+        // Non-streaming has no inter-token gaps: the body arrives whole.
+        tpot_p95_ms: None,
+        concurrency: None,
+        ttft_p95_s: None,
+        queue_wait_ms_median: None,
+        rejected: None,
+        artifact: None,
+        image_units: None,
+    })
+}
+
 /// Quality probe (#91): runs a fixed prompt and stores the full generated
 /// text as an artifact for later scoring (manual now, LLM-judge later). The
 /// point is to compare reasoning/planning quality across models — the axis
@@ -425,7 +655,7 @@ impl Scenario for CapabilityScenario {
             "stream": true,
             "stream_options": {"include_usage": true},
         });
-        let fut = stream_and_measure_inner(ctx, &payload, true);
+        let fut = stream_and_measure_inner(ctx, &payload, true, None);
         tokio::time::timeout(ctx.timeout, fut)
             .await
             .map_err(|_| anyhow!("capability probe timed out after {:?}", ctx.timeout))?
@@ -468,8 +698,9 @@ fn percentile(values: &[f64], p: f64) -> Option<f64> {
 async fn stream_and_measure(
     ctx: &RunCtx<'_>,
     payload: &serde_json::Value,
+    stream: Option<u32>,
 ) -> Result<ScenarioMetrics> {
-    stream_and_measure_inner(ctx, payload, false).await
+    stream_and_measure_inner(ctx, payload, false, stream).await
 }
 
 /// As [`stream_and_measure`] but accumulates the full visible text when
@@ -479,11 +710,10 @@ async fn stream_and_measure_inner(
     ctx: &RunCtx<'_>,
     payload: &serde_json::Value,
     capture_text: bool,
+    stream_ix: Option<u32>,
 ) -> Result<ScenarioMetrics> {
     let start = Instant::now();
-    let resp = ctx
-        .client
-        .post(&ctx.chat_url)
+    let resp = with_stream_identity(ctx.client.post(&ctx.chat_url), ctx, stream_ix)
         .json(payload)
         .send()
         .await
@@ -503,6 +733,10 @@ async fn stream_and_measure_inner(
     let mut prefill_ms: Option<u64> = None;
     let mut decode_ms: Option<u64> = None;
     let mut prefill_tokens: Option<u64> = None;
+    let mut reasoning_tokens: Option<u64> = None;
+    let mut cached_tokens: Option<u64> = None;
+    // Inter-token arrival gaps, for the p95 tail.
+    let mut gaps_ms: Vec<f64> = Vec::new();
     let mut captured = String::new();
 
     while let Some(event) = stream.next().await {
@@ -540,6 +774,11 @@ async fn stream_and_measure_inner(
             if content.is_some() || reasoning.is_some() {
                 if first.is_none() {
                     first = Some(now);
+                } else if let Some(prev) = last {
+                    // Gaps only *between* generated deltas: the wait for
+                    // the first one is TTFT and belongs to prefill, not
+                    // to stream smoothness.
+                    gaps_ms.push(now.duration_since(prev).as_secs_f64() * 1000.0);
                 }
                 last = Some(now);
                 chunk_count += 1;
@@ -551,6 +790,14 @@ async fn stream_and_measure_inner(
         if let Some(usage) = chunk.usage {
             prompt_tokens = Some(usage.prompt_tokens);
             completion_tokens = Some(usage.completion_tokens);
+            reasoning_tokens = usage
+                .completion_tokens_details
+                .as_ref()
+                .map(|d| d.reasoning_tokens);
+            cached_tokens = usage
+                .prompt_tokens_details
+                .as_ref()
+                .map(|d| d.cached_tokens);
             if let Some(t) = usage.helexa_timing {
                 prefill_ms = Some(t.prefill_ms);
                 decode_ms = Some(t.decode_ms);
@@ -589,6 +836,9 @@ async fn stream_and_measure_inner(
         prefill_ms,
         decode_ms,
         prefill_tokens,
+        reasoning_tokens,
+        cached_tokens,
+        tpot_p95_ms: percentile(&gaps_ms, 95.0),
         // Concurrency fields unset on the single-request path; the
         // concurrency scenario builds its own aggregate (#89).
         concurrency: None,
@@ -604,6 +854,67 @@ async fn stream_and_measure_inner(
 mod tests {
     use super::*;
 
+    /// #288/#54: a burst must present one identity *per stream*.
+    ///
+    /// Fair-share bounds a single principal to `max_per_principal`
+    /// in-flight-or-queued requests, which defaults to 2. Eight streams
+    /// under one key would have six rejected with `PrincipalCap`, and
+    /// `concurrency:8` would silently measure the fair-share cap instead
+    /// of the server — a worse failure than the anonymity it replaced,
+    /// because it looks like backpressure.
+    #[test]
+    fn a_burst_presents_one_identity_per_stream() {
+        let client = reqwest::Client::new();
+        let ctx = RunCtx {
+            client: &client,
+            chat_url: "http://x/v1/chat/completions".into(),
+            model_id: "m".into(),
+            max_tokens: 16,
+            timeout: Duration::from_secs(1),
+            principal_key_id: Some("fleet-benchmark".into()),
+        };
+        let key_of = |stream: Option<u32>| -> Option<String> {
+            with_stream_identity(client.post(&ctx.chat_url), &ctx, stream)
+                .build()
+                .expect("build")
+                .headers()
+                .get(HEADER_KEY_ID)
+                .map(|v| v.to_str().expect("utf8").to_string())
+        };
+        let keys: Vec<Option<String>> = (0..3).map(|i| key_of(Some(i))).collect();
+        assert_eq!(
+            keys,
+            vec![
+                Some("fleet-benchmark-0".to_string()),
+                Some("fleet-benchmark-1".to_string()),
+                Some("fleet-benchmark-2".to_string()),
+            ]
+        );
+        // Distinct, or the cap bites anyway.
+        let uniq: std::collections::HashSet<_> = keys.iter().collect();
+        assert_eq!(uniq.len(), 3);
+    }
+
+    /// A single-request scenario adds no override: the client's default
+    /// headers already carry the base identity, and a per-request copy
+    /// would be a second place for it to drift.
+    #[test]
+    fn a_single_request_scenario_keeps_the_client_identity() {
+        let client = reqwest::Client::new();
+        let ctx = RunCtx {
+            client: &client,
+            chat_url: "http://x/v1/chat/completions".into(),
+            model_id: "m".into(),
+            max_tokens: 16,
+            timeout: Duration::from_secs(1),
+            principal_key_id: Some("fleet-benchmark".into()),
+        };
+        let req = with_stream_identity(client.post(&ctx.chat_url), &ctx, None)
+            .build()
+            .expect("build");
+        assert!(req.headers().get(HEADER_KEY_ID).is_none());
+    }
+
     #[test]
     fn prompt_grows_with_token_target() {
         let small = build_prompt(128);
@@ -646,6 +957,7 @@ mod tests {
             prompt_sizes: vec![128],
             max_tokens: 64,
             concurrency_levels: vec![2, 8],
+            concurrency_nonstreaming_levels: vec![8],
             concurrency_prompt_tokens: 512,
             capability_probes: vec![CapabilityProbe {
                 name: "plan".into(),

crates/helexa-bench/src/store.rs

@@ -14,6 +14,22 @@ use std::path::Path;
 #[derive(Debug, Clone)]
 pub struct RunRecord {
     pub ts: String, // RFC3339
+    /// Identity this sample was taken under (#288), `account/key`, or
+    /// `None` for an anonymous run. Anonymous samples measure the
+    /// admission yield policy rather than serving capacity once #262 is
+    /// in the build, so this is the column that says whether two rows
+    /// are comparable at all.
+    pub principal: Option<String>,
+    /// Tokens spent reasoning (#223) — the dominant cost driver on a
+    /// reasoning model, and one that moves independently of every tok/s
+    /// number on the dashboard.
+    pub reasoning_tokens: Option<u64>,
+    /// Prompt tokens served from the prefix cache (#269) — the reason
+    /// prefill timing varies between otherwise identical samples.
+    pub cached_tokens: Option<u64>,
+    /// p95 inter-token arrival gap (ms) — stream smoothness, which a
+    /// mean tok/s cannot express.
+    pub tpot_p95_ms: Option<f64>,
     // target
     pub target_name: String,
     pub target_kind: String,
@@ -204,6 +220,18 @@ impl Store {
                 ("artifact", "TEXT"),
                 ("quality_score", "REAL"),
                 ("scorer", "TEXT"),
+                // #288: which identity the sample was taken under. Rows
+                // predating this are NULL, which is honest — they were
+                // anonymous, and since #262 that measures a different
+                // thing. Without the column the 2026-08-18 discontinuity
+                // is folklore rather than data.
+                ("principal", "TEXT"),
+                // #223 / #269: both are on the wire and were being discarded.
+                ("reasoning_tokens", "INTEGER"),
+                ("cached_tokens", "INTEGER"),
+                // #285: inter-token tail. `decode_tps` is a mean and hides
+                // a stall that later catches up.
+                ("tpot_p95_ms", "REAL"),
             ],
         )?;
         Ok(())
@@ -264,6 +292,7 @@ impl Store {
                 swap_unload_ms, swap_load_ms,
                 artifact, quality_score, scorer,
                 image_units,
+                principal, reasoning_tokens, cached_tokens, tpot_p95_ms,
                 ok, error
             ) VALUES (
                 ?1, ?2, ?3, ?4,
@@ -280,7 +309,8 @@ impl Store {
                 ?42, ?43,
                 ?44, ?45, ?46,
                 ?47,
-                ?48, ?49
+                ?48, ?49, ?50, ?51,
+                ?52, ?53
             )",
             params![
                 r.ts,
@@ -330,6 +360,10 @@ impl Store {
                 r.quality_score,
                 r.scorer,
                 r.image_units,
+                r.principal,
+                r.reasoning_tokens,
+                r.cached_tokens,
+                r.tpot_p95_ms,
                 r.ok as i64,
                 r.error,
             ],
@@ -703,7 +737,15 @@ impl Store {
             }
         };
         let mut stmt = self.conn.prepare(
-            "SELECT git_sha, build_timestamp, package_version, ttft_s, decode_tps, total_s, ts
+            "SELECT git_sha, build_timestamp, package_version, ttft_s, decode_tps, total_s,
+                    ttft_p95_s, queue_wait_ms, rejected,
+                    -- Prefill rate derived here rather than stored: the
+                    -- inputs are already recorded and a stored rate
+                    -- would be a third thing to keep consistent.
+                    CASE WHEN prefill_ms > 0 AND prefill_tokens IS NOT NULL
+                         THEN (prefill_tokens * 1000.0) / prefill_ms END AS prefill_tps,
+                    reasoning_tokens, cached_tokens, completion_tokens,
+                    tpot_p95_ms, principal, ts
              FROM runs
              WHERE ok=1 AND target_name=?1 AND model_id=?2 AND scenario_id=?3
              ORDER BY id",
@@ -717,7 +759,16 @@ impl Store {
                     ttft_s: r.get(3)?,
                     decode_tps: r.get(4)?,
                     total_s: r.get(5)?,
-                    ts: r.get(6)?,
+                    ttft_p95_s: r.get(6)?,
+                    queue_wait_ms: r.get(7)?,
+                    rejected: r.get(8)?,
+                    prefill_tps: r.get(9)?,
+                    reasoning_tokens: r.get(10)?,
+                    cached_tokens: r.get(11)?,
+                    completion_tokens: r.get(12)?,
+                    tpot_p95_ms: r.get(13)?,
+                    principal: r.get(14)?,
+                    ts: r.get(15)?,
                 })
             })?
             .collect::<rusqlite::Result<_>>()?;
@@ -838,6 +889,33 @@ pub struct SeriesPoint {
     pub ttft_s_median: Option<f64>,
     pub decode_tps_median: Option<f64>,
     pub total_s_median: Option<f64>,
+    // ── Added because the two above cannot show what changed ──────
+    /// Tail TTFT. The median above stayed flat at ~0.3 s across the
+    /// build where p95 went 0.53 s → 11.77 s (#288); charting only the
+    /// median is why that went unnoticed for a week.
+    pub ttft_p95_s_median: Option<f64>,
+    /// Admission queue wait — separates "the server is slow" from
+    /// "you were queued behind someone".
+    pub queue_wait_ms_median: Option<f64>,
+    /// Requests shed by admission during the burst.
+    pub rejected_median: Option<f64>,
+    /// Prefill throughput, derived from `prefill_tokens / prefill_ms`.
+    /// The other half of serving speed; only decode was ever charted.
+    pub prefill_tps_median: Option<f64>,
+    /// Tokens spent reasoning (#223) — the dominant cost driver on a
+    /// reasoning model, and independent of every speed metric here.
+    pub reasoning_tokens_median: Option<f64>,
+    /// Prompt tokens served from the prefix cache (#269).
+    pub cached_tokens_median: Option<f64>,
+    /// Total generated tokens — catches a model that starts truncating
+    /// or rambling, which no rate metric shows.
+    pub completion_tokens_median: Option<f64>,
+    /// p95 inter-token gap (ms) — stream smoothness.
+    pub tpot_p95_ms_median: Option<f64>,
+    /// Identity the samples were taken under (#288), or `None` if
+    /// anonymous. Anonymous and identified points are **not
+    /// comparable** once #262 is in the build.
+    pub principal: Option<String>,
     pub samples: usize,
 }
 
@@ -848,6 +926,15 @@ struct SeriesRaw {
     ttft_s: Option<f64>,
     decode_tps: Option<f64>,
     total_s: Option<f64>,
+    ttft_p95_s: Option<f64>,
+    queue_wait_ms: Option<f64>,
+    rejected: Option<f64>,
+    prefill_tps: Option<f64>,
+    reasoning_tokens: Option<f64>,
+    cached_tokens: Option<f64>,
+    completion_tokens: Option<f64>,
+    tpot_p95_ms: Option<f64>,
+    principal: Option<String>,
     ts: String,
 }
 
@@ -880,6 +967,19 @@ fn aggregate_series(raws: Vec<SeriesRaw>) -> Vec<SeriesPoint> {
                 ttft_s_median: median(rows.iter().filter_map(|r| r.ttft_s)),
                 decode_tps_median: median(rows.iter().filter_map(|r| r.decode_tps)),
                 total_s_median: median(rows.iter().filter_map(|r| r.total_s)),
+                ttft_p95_s_median: median(rows.iter().filter_map(|r| r.ttft_p95_s)),
+                queue_wait_ms_median: median(rows.iter().filter_map(|r| r.queue_wait_ms)),
+                rejected_median: median(rows.iter().filter_map(|r| r.rejected)),
+                prefill_tps_median: median(rows.iter().filter_map(|r| r.prefill_tps)),
+                reasoning_tokens_median: median(rows.iter().filter_map(|r| r.reasoning_tokens)),
+                cached_tokens_median: median(rows.iter().filter_map(|r| r.cached_tokens)),
+                completion_tokens_median: median(rows.iter().filter_map(|r| r.completion_tokens)),
+                tpot_p95_ms_median: median(rows.iter().filter_map(|r| r.tpot_p95_ms)),
+                // Any row's principal: a build's samples are all taken
+                // under one identity, and a mixture would mean the
+                // config changed mid-build, which the chart should show
+                // as a discontinuity rather than average away.
+                principal: rows.iter().find_map(|r| r.principal.clone()),
                 samples: rows.len(),
             };
             (sort_key, point)
@@ -1281,6 +1381,10 @@ mod tests {
     fn rec(target: &str, sha: &str, model: &str, scenario: &str, ok: bool) -> RunRecord {
         RunRecord {
             ts: "2026-06-13T00:00:00Z".into(),
+            principal: None,
+            reasoning_tokens: None,
+            cached_tokens: None,
+            tpot_p95_ms: None,
             target_name: target.into(),
             target_kind: "neuron".into(),
             endpoint: "http://x:13131".into(),

crates/helexa-bench/src/sweep.rs

@@ -78,7 +78,10 @@ pub struct Sweeper {
 
 impl Sweeper {
     pub fn new(cfg: BenchConfig) -> Result<Self> {
-        let client = TargetClient::new(cfg.bench.request_timeout())?;
+        let client = TargetClient::with_principal(
+            cfg.bench.request_timeout(),
+            cfg.bench.principal.as_ref(),
+        )?;
         let store = Store::open(&cfg.bench.db_path)?;
         Ok(Sweeper { cfg, client, store })
     }
@@ -169,6 +172,7 @@ impl Sweeper {
             model_id: model.id.clone(),
             max_tokens: self.cfg.scenarios.max_tokens,
             timeout: self.cfg.bench.request_timeout(),
+            principal_key_id: self.cfg.bench.principal.as_ref().map(|p| p.key_id.clone()),
         };
         let cold = crate::scenario::cold_probe(&ctx).await;
         let swap = SwapTiming { unload_ms, load_ms };
@@ -252,6 +256,7 @@ impl Sweeper {
                     model_id: model.id.clone(),
                     max_tokens: self.cfg.scenarios.max_tokens,
                     timeout: self.cfg.bench.request_timeout(),
+                    principal_key_id: self.cfg.bench.principal.as_ref().map(|p| p.key_id.clone()),
                 };
 
                 // One unmeasured warmup when the cell is empty (matches
@@ -346,6 +351,15 @@ impl Sweeper {
 
         RunRecord {
             ts: chrono::Utc::now().to_rfc3339(),
+            // #288: stamp the identity this sample was taken under, so a
+            // later reader can tell an anonymous row from an identified
+            // one instead of inferring it from the date.
+            principal: self
+                .cfg
+                .bench
+                .principal
+                .as_ref()
+                .map(|p| format!("{}/{}", p.account_id, p.key_id)),
             target_name: target.name.clone(),
             target_kind: kind_str(target.kind).to_string(),
             endpoint: target.endpoint.clone(),
@@ -383,6 +397,9 @@ impl Sweeper {
             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),
+            reasoning_tokens: m.and_then(|x| x.reasoning_tokens),
+            cached_tokens: m.and_then(|x| x.cached_tokens),
+            tpot_p95_ms: m.and_then(|x| x.tpot_p95_ms),
             vram_used_mb: health.map(|h| h.vram_used_mb),
             gpu_util_pct: health.map(|h| h.gpu_util_pct),
             gpu_temp_c: health.map(|h| h.gpu_temp_c),

crates/helexa-bench/tests/api.rs

@@ -15,6 +15,10 @@ fn rec(
     ok: bool,
 ) -> RunRecord {
     RunRecord {
+        principal: None,
+        reasoning_tokens: None,
+        cached_tokens: None,
+        tpot_p95_ms: None,
         ts: "2026-06-13T00:00:00Z".into(),
         target_name: host.into(),
         target_kind: "neuron".into(),

crates/helexa-bench/tests/sweep_integration.rs

@@ -80,6 +80,7 @@ async fn spawn_mock(sha: &str) -> (String, Arc<Mutex<String>>) {
 fn config_for(endpoint: String, db_path: String) -> BenchConfig {
     BenchConfig {
         bench: BenchSettings {
+            principal: None,
             sweep_interval_secs: 1,
             samples_per_version: 2,
             iteration_pause_secs: 0,
@@ -91,6 +92,7 @@ fn config_for(endpoint: String, db_path: String) -> BenchConfig {
             prompt_sizes: vec![128], // single scenario keeps assertions simple
             max_tokens: 16,
             concurrency_levels: Vec::new(),
+            concurrency_nonstreaming_levels: vec![],
             concurrency_prompt_tokens: 512,
             capability_probes: Vec::new(),
         },