feat(bench): concurrency-sweep knee report + /api/concurrency (#137 T3)

Turns the #89 concurrency-burst data into a capacity verdict. Adds
Store::concurrency() — per (target, model) at the latest build, the
throughput / p95-TTFT / queue-wait / reject-rate curve across burst
widths, plus the derived knee: the highest concurrency the model
sustains before it sheds load or its p95 TTFT tail stretches past 2× the
lightest-load baseline. This is the data-backed justification for a
max_in_flight ceiling.

Surfaced three ways:
- `helexa-bench report --concurrency` (md/json)
- GET /api/concurrency — the bench UI's live-capacity view
- example config enables the sweep [1,2,4,8,16] to bracket beast's
  max_in_flight=8 so the knee is visible

Tests: the knee lands on the last sustainable level; None when even the
lightest level sheds.

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

crates/helexa-bench/src/api.rs

@@ -36,6 +36,7 @@ pub fn api_routes(state: ApiState) -> Router {
         .route("/api/dimensions", get(dimensions))
         .route("/api/summary", get(summary))
         .route("/api/scaling", get(scaling))
+        .route("/api/concurrency", get(concurrency))
         .route("/api/swap", get(swap))
         .route("/api/capability", get(capability))
         .route("/api/series", get(series))
@@ -85,6 +86,16 @@ async fn scaling(
     store.scaling().map(Json).map_err(err500)
 }
 
+/// Concurrency sweep per (target, model) — throughput / p95 TTFT / reject-rate
+/// across burst widths, plus the sustainable-concurrency knee (#137). The UI's
+/// live-capacity view.
+async fn concurrency(
+    State(s): State<ApiState>,
+) -> Result<Json<Vec<crate::store::ConcurrencyCurve>>, ApiError> {
+    let store = s.lock().await;
+    store.concurrency().map(Json).map_err(err500)
+}
+
 /// Cold-load / model-swap costs per (target, model) — reload latency + cold
 /// first-request (#90).
 async fn swap(State(s): State<ApiState>) -> Result<Json<Vec<crate::store::SwapCost>>, ApiError> {

crates/helexa-bench/src/main.rs

@@ -94,6 +94,11 @@ enum Command {
         /// scores, with per-model median.
         #[arg(long)]
         capability: bool,
+        /// Render the concurrency-sweep view (#137): throughput / p95 TTFT /
+        /// reject-rate across burst widths per model, with the sustainable-
+        /// concurrency knee.
+        #[arg(long)]
+        concurrency: bool,
     },
 }
 
@@ -205,6 +210,7 @@ async fn run(cli: Cli) -> Result<()> {
             scaling,
             swap,
             capability,
+            concurrency,
         } => {
             let db_path = match db {
                 Some(p) => p,
@@ -217,6 +223,12 @@ async fn run(cli: Cli) -> Result<()> {
                     Format::Md => report::render_capability_markdown(&runs),
                     Format::Json => report::render_capability_json(&runs)?,
                 }
+            } else if concurrency {
+                let curves = store.concurrency()?;
+                match format {
+                    Format::Md => report::render_concurrency_markdown(&curves),
+                    Format::Json => report::render_concurrency_json(&curves)?,
+                }
             } else if swap {
                 let costs = store.swap_costs()?;
                 match format {

crates/helexa-bench/src/report.rs

@@ -3,7 +3,7 @@
 //! doc: engine, model, prompt tok, TTFT (s), decode tok/s, total (s),
 //! plus the build SHA each cell was measured against.
 
-use crate::store::{CapabilityRun, ReportRow, ScalingCurve, SwapCost};
+use crate::store::{CapabilityRun, ConcurrencyCurve, ReportRow, ScalingCurve, SwapCost};
 use anyhow::Result;
 
 pub fn render_markdown(rows: &[ReportRow]) -> String {
@@ -136,6 +136,58 @@ pub fn render_scaling_json(curves: &[ScalingCurve]) -> Result<String> {
     Ok(serde_json::to_string_pretty(curves)?)
 }
 
+/// Concurrency-sweep view (#137): one block per (target, model) with the
+/// throughput / latency-tail / shedding curve across burst widths, then the
+/// knee — the max sustainable concurrency, the data-backed `max_in_flight`.
+pub fn render_concurrency_markdown(curves: &[ConcurrencyCurve]) -> String {
+    let mut out = String::new();
+    for c in curves {
+        let gpu = c.gpu.as_deref().unwrap_or("");
+        out.push_str(&format!(
+            "### {} · {}  (`{}`{})\n\n",
+            c.target_name,
+            c.model_id,
+            c.git_sha,
+            if gpu.is_empty() {
+                String::new()
+            } else {
+                format!(", {gpu}")
+            },
+        ));
+        out.push_str("| N | decode tok/s | p95 TTFT (s) | queue wait (ms) | reject % | n |\n");
+        out.push_str("|---:|---:|---:|---:|---:|---:|\n");
+        for p in &c.points {
+            let reject = p
+                .reject_rate
+                .map(|r| format!("{:.0}%", r * 100.0))
+                .unwrap_or_else(|| "—".into());
+            out.push_str(&format!(
+                "| {} | {} | {} | {} | {} | {} |\n",
+                p.concurrency,
+                fmt_opt(p.decode_tps, 1),
+                fmt_opt(p.ttft_p95_s, 2),
+                fmt_opt(p.queue_wait_ms, 0),
+                reject,
+                p.samples,
+            ));
+        }
+        match c.knee_concurrency {
+            Some(k) => out.push_str(&format!(
+                "\nmax sustainable concurrency: **{k}** \
+                 (no shedding, p95 TTFT within 2× of the lightest-load baseline)\n\n",
+            )),
+            None => out.push_str(
+                "\nmax sustainable concurrency: — (sheds or breaks even at the lightest level)\n\n",
+            ),
+        }
+    }
+    out
+}
+
+pub fn render_concurrency_json(curves: &[ConcurrencyCurve]) -> Result<String> {
+    Ok(serde_json::to_string_pretty(curves)?)
+}
+
 /// Cold-load / model-swap cost view (#90): reload latency + cold
 /// first-request per model.
 pub fn render_swap_markdown(costs: &[SwapCost]) -> String {

crates/helexa-bench/src/store.rs

@@ -431,6 +431,58 @@ impl Store {
         Ok(out)
     }
 
+    /// Concurrency sweep (#137): per (target, model) at the latest build, the
+    /// aggregate serving behaviour across `concurrency:<n>` burst levels, plus
+    /// the derived knee — the max sustainable concurrency before the model
+    /// sheds load or its p95 TTFT tail breaks. The data-backed justification
+    /// for a `max_in_flight` ceiling.
+    pub fn concurrency(&self) -> Result<Vec<ConcurrencyCurve>> {
+        use std::collections::BTreeMap;
+        // Reuse the aggregated report cells; the concurrency:<n> rows are the
+        // per-burst-width measurement points.
+        let mut by_model: BTreeMap<(String, String), Vec<ReportRow>> = BTreeMap::new();
+        for r in self.report_rows()? {
+            if r.scenario_id.starts_with("concurrency:") {
+                by_model
+                    .entry((r.target_name.clone(), r.model_id.clone()))
+                    .or_default()
+                    .push(r);
+            }
+        }
+        let mut out = Vec::new();
+        for ((target_name, model_id), mut rows) in by_model {
+            rows.sort_by_key(|r| r.concurrency.unwrap_or(0));
+            let points: Vec<ConcurrencyPoint> = rows
+                .iter()
+                .map(|r| {
+                    let n = r.concurrency.unwrap_or(0);
+                    let reject_rate = match r.rejected_median {
+                        Some(rej) if n > 0 => Some(rej / n as f64),
+                        _ => None,
+                    };
+                    ConcurrencyPoint {
+                        concurrency: n as u32,
+                        decode_tps: r.decode_tps_median,
+                        ttft_p95_s: r.ttft_p95_load_s,
+                        queue_wait_ms: r.queue_wait_ms_median,
+                        reject_rate,
+                        samples: r.samples,
+                    }
+                })
+                .collect();
+            let knee_concurrency = concurrency_knee(&points);
+            out.push(ConcurrencyCurve {
+                target_name,
+                model_id,
+                git_sha: rows.first().map(|r| r.git_sha.clone()).unwrap_or_default(),
+                gpu: rows.iter().find_map(|r| r.gpu.clone()),
+                points,
+                knee_concurrency,
+            });
+        }
+        Ok(out)
+    }
+
     /// Cold-load / model-swap costs (#90): per (target, model) at the latest
     /// build, the median unload→reload latency and the cold first-request
     /// latency after reload (the `scenario_id = "swap"` rows).
@@ -1007,6 +1059,68 @@ pub struct ScalingPoint {
     pub samples: usize,
 }
 
+/// One point on a concurrency sweep for a (target, model) at its latest
+/// build (#137): the aggregate serving behaviour at `concurrency` N
+/// simultaneous streams.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct ConcurrencyPoint {
+    /// Burst width — the number of simultaneous streams.
+    pub concurrency: u32,
+    /// Aggregate node throughput (total tokens / burst window), tok/s.
+    pub decode_tps: Option<f64>,
+    /// Within-burst p95 time-to-first-token — where saturation first bites.
+    pub ttft_p95_s: Option<f64>,
+    /// Median admission queue wait, ms.
+    pub queue_wait_ms: Option<f64>,
+    /// Fraction of the burst shed by admission (`rejected / concurrency`).
+    /// `> 0` means the model is at capacity for this N.
+    pub reject_rate: Option<f64>,
+    pub samples: usize,
+}
+
+/// A concurrency sweep for a (target, model) at its latest build (#137):
+/// the points ordered by burst width, plus the derived knee — the highest
+/// concurrency the model sustains before it sheds load or its latency tail
+/// breaks. This is the data-backed justification for a `max_in_flight`
+/// ceiling.
+#[derive(Debug, Clone, serde::Serialize)]
+pub struct ConcurrencyCurve {
+    pub target_name: String,
+    pub model_id: String,
+    pub git_sha: String,
+    pub gpu: Option<String>,
+    pub points: Vec<ConcurrencyPoint>,
+    /// Max sustainable concurrency: the largest level that shed nothing and
+    /// held p95 TTFT within [`KNEE_TTFT_FACTOR`] of the lightest-load
+    /// baseline. `None` when even the smallest level sheds or lacks a TTFT.
+    pub knee_concurrency: Option<u32>,
+}
+
+/// How far the p95 TTFT tail may stretch past the lightest-load baseline
+/// before a concurrency level counts as "broken" for the knee (#137).
+const KNEE_TTFT_FACTOR: f64 = 2.0;
+
+/// Highest sustainable concurrency from an ascending-ordered sweep: the
+/// largest level that shed nothing and held p95 TTFT within
+/// [`KNEE_TTFT_FACTOR`] of the lightest-load baseline. Degradation is
+/// monotonic in N, so the scan stops at the first level that breaks.
+fn concurrency_knee(points: &[ConcurrencyPoint]) -> Option<u32> {
+    let baseline = points.iter().find_map(|p| p.ttft_p95_s)?;
+    let mut knee = None;
+    for p in points {
+        let shed = p.reject_rate.unwrap_or(0.0) > 0.0;
+        let within_tail = p
+            .ttft_p95_s
+            .map(|t| t <= baseline * KNEE_TTFT_FACTOR)
+            .unwrap_or(false);
+        if shed || !within_tail {
+            break;
+        }
+        knee = Some(p.concurrency);
+    }
+    knee
+}
+
 /// Group by (target, model, scenario), keep only the latest SHA's rows
 /// (latest = the SHA of the last-inserted row, since input is id-ordered),
 /// and median each metric.
@@ -1313,6 +1427,44 @@ mod tests {
         assert_eq!(row.ttft_p95_load_s, Some(0.9));
     }
 
+    #[test]
+    fn concurrency_sweep_finds_the_knee() {
+        let s = Store::open_in_memory().unwrap();
+        // A sweep where 1,2,4 hold but 8 sheds and blows the p95 tail.
+        for (n, ttft, rej) in [(1u32, 0.2, 0u32), (2, 0.25, 0), (4, 0.35, 0), (8, 0.9, 2)] {
+            let mut r = rec("beast", "sha", "m", &format!("concurrency:{n}"), true);
+            r.concurrency = Some(n);
+            r.ttft_p95_s = Some(ttft);
+            r.rejected = Some(rej);
+            s.insert_run(&r).unwrap();
+        }
+        let curves = s.concurrency().unwrap();
+        assert_eq!(curves.len(), 1);
+        let c = &curves[0];
+        // Ordered ascending by burst width.
+        assert_eq!(
+            c.points.iter().map(|p| p.concurrency).collect::<Vec<_>>(),
+            vec![1, 2, 4, 8]
+        );
+        // N=8 shed 2 of 8 → reject_rate 0.25.
+        assert_eq!(c.points[3].reject_rate, Some(0.25));
+        // Knee = 4: highest level with no shedding and p95 TTFT within 2× of
+        // the lightest-load baseline (0.2 → 0.4; 0.35 holds, 0.9 breaks).
+        assert_eq!(c.knee_concurrency, Some(4));
+    }
+
+    #[test]
+    fn concurrency_knee_is_none_when_lightest_level_sheds() {
+        let s = Store::open_in_memory().unwrap();
+        let mut r = rec("beast", "sha", "m", "concurrency:2", true);
+        r.concurrency = Some(2);
+        r.ttft_p95_s = Some(0.3);
+        r.rejected = Some(1);
+        s.insert_run(&r).unwrap();
+        let curves = s.concurrency().unwrap();
+        assert_eq!(curves[0].knee_concurrency, None);
+    }
+
     #[test]
     fn capability_runs_store_artifact_and_accept_scores() {
         let s = Store::open_in_memory().unwrap();

helexa-bench.example.toml

@@ -32,10 +32,13 @@ prompt_sizes = [128, 4096]
 max_tokens = 256
 # Concurrency / agentic-load scenarios (#89): one concurrency:<n> scenario
 # per level, each firing N simultaneous streams to characterize the real
-# a0/hermes/opencode fan-out. Empty by default — enable deliberately, since
-# a burst puts genuine simultaneous load on the serving fleet.
-# concurrency_levels = [2, 4, 8]
-# concurrency_prompt_tokens = 512
+# a0/hermes/opencode fan-out. Enable deliberately — a burst puts genuine
+# simultaneous load on the serving fleet. Bracket the model's max_in_flight
+# (beast serves 8) so `helexa-bench report --concurrency` (or GET
+# /api/concurrency) can locate the sustainable-concurrency knee (#137): the
+# highest N before the model sheds load or its p95 TTFT tail breaks.
+concurrency_levels = [1, 2, 4, 8, 16]
+concurrency_prompt_tokens = 512
 
 # Capability probes (#91): each runs a fixed prompt and stores the full
 # output for quality scoring — the reasoning/planning axis the speed