metrale_bench/benchmarks/serve_matrix/
mod.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: Serve Matrix: boot every checkpoint the box can serve, run the
4//! coherence, codegen, tool-call, long-context and throughput probes on it, and
5//! score one row per round.
6//!
7//! Owner: bench, serve matrix.
8//! Invariants:
9//! - The rounds come from the host's roster (`ServeHost::roster`); this crate
10//!   lists no model.
11//! - Scoring walks the plan, not the results, so a planned round with no result
12//!   fails (`score::tally`).
13
14pub mod host;
15pub mod plan;
16pub mod probes;
17pub mod report;
18pub mod schema;
19pub mod score;
20
21use crate::hardware::Sensitivity;
22use std::future::Future;
23use std::sync::Arc;
24use std::time::{Duration, Instant};
25
26use anyhow::{Context, Result};
27
28use crate::benchmark::{Benchmark, BenchmarkDescriptor};
29use crate::benchmarks::{baseline, one_line, unique_prefix_tag};
30use crate::http;
31use crate::metadata::PluginMetadata;
32use crate::params::{ParamSpec, ParamValues};
33use crate::plugin::{Plugin, PluginHandle};
34use crate::result::{BenchmarkResult, LogLine, RunStatus};
35
36use host::{ServeHost, ServeOptions};
37use plan::{Plan, Round};
38use score::{Outcome, RoundResult, Signals};
39
40/// 2026-09-26: Room `configure` reserves beyond the long-context filler for the
41/// needle sentence, the question, the chat template and the probe's 32-token
42/// answer budget.
43const LONG_CTX_HEADROOM: usize = 256;
44
45const SUMMARY: &str = "Boot every servable checkpoint, probe it, gate the matrix";
46pub const METADATA: PluginMetadata = PluginMetadata::metrale(SUMMARY);
47
48pub const DESCRIPTOR: BenchmarkDescriptor = BenchmarkDescriptor {
49    id: "serve-matrix",
50    name: "Serve Matrix",
51    summary: SUMMARY,
52    detail: "The release serve gate. For every checkpoint this box can actually serve — cached \
53             weights intersected with the kernels this build compiled — it loads the model, \
54             checks coherence, code generation, tool calling and long-context recall, measures \
55             decode throughput, and moves on. Coverage is enforced: a checkpoint that is \
56             planned and fails to boot is a FAIL, never a silent absence. Booting replaces \
57             whatever model is currently serving; the one that was serving when the run started \
58             is restored at the end.",
59    duration_hint: "~5–10 min per checkpoint",
60    expected_secs: 600,
61    updated: "2026-08-02",
62    // 2026-09-26: Every round swaps out the model the box is serving.
63    needs_confirmation: true,
64    // 2026-09-26: The matrix covers whatever the box holds.
65    intended_for: None,
66    threshold_params: &[],
67    // 2026-09-26: Correctness, so the hardware state is recorded and does not
68    // gate the run. The bars still include tok/s: zero, non-finite, or more
69    // than `score::TPS_TOLERANCE` below a stored baseline fails
70    // (`RoundResult::bars`).
71    sensitivity: Sensitivity::Correctness,
72    ctor: || Box::new(ServeMatrix::default()),
73};
74
75#[derive(Default)]
76pub struct ServeMatrix {
77    handle: Option<PluginHandle>,
78    host: Option<Arc<dyn ServeHost>>,
79    started: Option<Instant>,
80    include: String,
81    opts: Option<ServeOptions>,
82    long_ctx_tokens: usize,
83    tps_tokens: usize,
84    probe_budget: usize,
85    timeout: Duration,
86    update_baselines: bool,
87    /// 2026-09-26: The roster, classified. The cursor indexes `plan.planned()`,
88    /// the iterator the scoring walks.
89    plan: Plan,
90    planned_built: bool,
91    cursor: usize,
92    results: Vec<RoundResult>,
93    baseline: Option<baseline::Baseline>,
94}
95
96impl ServeMatrix {
97    /// 2026-09-26: Construct against an explicit host. Tests use this; the
98    /// registry's `ctor` takes no arguments.
99    pub fn with_host(host: Arc<dyn ServeHost>) -> Self {
100        Self {
101            host: Some(host),
102            ..Self::default()
103        }
104    }
105
106    fn handle(&self) -> Result<&PluginHandle> {
107        self.handle.as_ref().context("benchmark was not loaded")
108    }
109
110    fn host(&self) -> Result<Arc<dyn ServeHost>> {
111        self.host.clone().context(host::NO_HOST)
112    }
113
114    fn options(&self) -> Result<ServeOptions> {
115        self.opts.context("benchmark was not configured")
116    }
117
118    fn elapsed(&self) -> Duration {
119        self.started.map(|s| s.elapsed()).unwrap_or_default()
120    }
121
122    fn baseline_for(&self, label: &str) -> Option<f64> {
123        self.baseline.as_ref()?.get(&format!("tps:{label}"))
124    }
125
126    /// 2026-09-26: Boot one checkpoint and probe it. Returns an `Outcome`, never
127    /// an error: a failed round is a row and the matrix moves on.
128    async fn run_round(&self, round: &Round) -> Outcome {
129        let Ok(handle) = self.handle().cloned() else {
130            return Outcome::NotReached;
131        };
132        let (Ok(host), Ok(opts)) = (self.host(), self.options()) else {
133            return Outcome::NotReached;
134        };
135        handle.status(format!("{} · booting", round.label()));
136        let target = match host.serve(&round.model, opts).await {
137            Ok(t) => t,
138            Err(e) => return Outcome::BootFailed(one_line(format!("{e:#}"))),
139        };
140        // 2026-09-26: `ServeHost::serve` promises an answering endpoint; this
141        // probe checks it, so a host that returned early gives a boot failure.
142        if let Err(e) = http::probe(&target, Duration::from_secs(30)).await {
143            return Outcome::BootFailed(format!(
144                "came up but did not answer: {}",
145                one_line(format!("{e:#}"))
146            ));
147        }
148
149        let mut s = Signals::default();
150        handle.status(format!("{} · coherence", round.label()));
151        let (coherence, concern) = probes::coherence_probe(&target, self.timeout).await;
152        s.coherence_pass = coherence.passed;
153        s.coherence_total = coherence.total;
154        s.identity = coherence.identity;
155        if !concern.is_empty() {
156            handle.warn(format!("{}: {concern}", round.label()));
157        }
158
159        handle.status(format!("{} · codegen", round.label()));
160        s.codegen = probes::codegen_probe(&target, self.timeout, self.probe_budget).await;
161        handle.status(format!("{} · tool call", round.label()));
162        s.tool_call = probes::tool_call_probe(&target, self.timeout, self.probe_budget).await;
163
164        if self.long_ctx_tokens > 0 {
165            handle.status(format!("{} · long context", round.label()));
166            let tag = unique_prefix_tag(&format!("sm-lc-{}", self.cursor), handle.run_id());
167            s.long_ctx =
168                probes::long_context_probe(&target, self.timeout, self.long_ctx_tokens, &tag).await;
169        }
170
171        handle.status(format!("{} · throughput", round.label()));
172        let tag = unique_prefix_tag(&format!("sm-tps-{}", self.cursor), handle.run_id());
173        let (tps, err) = probes::tps_probe(&target, self.timeout, self.tps_tokens, &tag).await;
174        s.tps = tps;
175        if let Some(e) = err {
176            handle.warn(format!("{}: throughput probe failed: {e}", round.label()));
177        }
178        Outcome::Probed(Box::new(s))
179    }
180
181    /// 2026-09-26: Ask the host for the box's roster and classify it.
182    fn build_plan(&mut self) -> Result<()> {
183        let roster = self.host()?.roster().context("reading the box's roster")?;
184        self.plan = Plan::build(&roster, &self.include);
185        self.planned_built = true;
186        Ok(())
187    }
188
189    /// 2026-09-26: Store the measured tok/s as the new baseline. Called only
190    /// when `update_baselines` is set, and takes only rounds with no failed bar
191    /// and a positive tok/s.
192    fn write_baselines(&self) -> Result<usize> {
193        let handle = self.handle()?;
194        let mut metrics = std::collections::BTreeMap::new();
195        for r in self.results.iter().filter(|r| r.bars().is_empty()) {
196            if let Some(tps) = r.signals().and_then(|s| s.tps).filter(|v| *v > 0.0) {
197                metrics.insert(format!("tps:{}", r.label), (tps * 10.0).round() / 10.0);
198            }
199        }
200        let written = metrics.len();
201        baseline::save(
202            handle.artifacts(),
203            DESCRIPTOR.id,
204            &handle.target().base_url,
205            &handle.target().model,
206            metrics,
207        )?;
208        Ok(written)
209    }
210
211    fn finish(&self) -> Result<BenchmarkResult> {
212        let tally = score::tally(&self.plan, &self.results);
213        let total = self.plan.planned_count() as u64;
214        let mut frame = BenchmarkResult {
215            status: RunStatus::Completed,
216            ..BenchmarkResult::running("done", self.elapsed())
217        }
218        .with_progress(total, total)
219        .with_summary(report::summary(&tally))
220        .with_table(report::table(&self.plan, &self.results))
221        .with_verdict(report::verdict(&tally, &self.plan));
222        if self.update_baselines {
223            match self.write_baselines() {
224                Ok(n) => {
225                    frame = frame.log_line(LogLine::info(format!(
226                        "{n} throughput baseline(s) refreshed from the rounds that PASSED — later \
227                         runs gate against these. Rounds below bar were not blessed."
228                    )))
229                }
230                Err(e) => {
231                    frame = frame.log_line(LogLine::warn(format!("baselines not written: {e:#}")))
232                }
233            }
234        } else if self.baseline.is_none() {
235            frame = frame.log_line(LogLine::warn(
236                "no throughput baseline on this box: the tok/s column is liveness only, not a \
237                 regression check. Re-run with `update baselines` on to record one.",
238            ));
239        }
240        Ok(frame)
241    }
242}
243
244impl Plugin for ServeMatrix {
245    fn metadata(&self) -> &'static PluginMetadata {
246        &METADATA
247    }
248
249    fn load(&mut self, handle: PluginHandle) -> impl Future<Output = Result<()>> + Send {
250        // 2026-09-26: A host passed to `with_host` wins; otherwise the installed
251        // one. With neither, `load` fails with `NO_HOST`.
252        if self.host.is_none() {
253            self.host = host::installed();
254        }
255        self.baseline = baseline::load(handle.artifacts(), DESCRIPTOR.id);
256        self.handle = Some(handle);
257        self.started = Some(Instant::now());
258        let ok = self.host.is_some();
259        async move {
260            if ok {
261                Ok(())
262            } else {
263                anyhow::bail!(host::NO_HOST)
264            }
265        }
266    }
267}
268
269impl Benchmark for ServeMatrix {
270    fn descriptor(&self) -> &'static BenchmarkDescriptor {
271        &DESCRIPTOR
272    }
273
274    fn parameters(&self) -> Vec<ParamSpec> {
275        schema::specs()
276    }
277
278    fn configure(&mut self, values: &ParamValues) -> Result<()> {
279        let specs = self.parameters();
280        values.validate_against(&specs)?;
281        let include = values.text("include")?.trim().to_string();
282        // 2026-09-26: `all` is the schema's word for "no filter", because
283        // `ParamKind::Text` refuses an empty value.
284        self.include = if include.eq_ignore_ascii_case("all") {
285            String::new()
286        } else {
287            include
288        };
289        let max_seq_len = values.usize("max_seq_len")?;
290        self.long_ctx_tokens = values.usize("long_ctx_tokens")?;
291        if self.long_ctx_tokens + LONG_CTX_HEADROOM > max_seq_len {
292            // 2026-09-26: The prompt is the filler plus the needle and the
293            // question, and the reply needs room too, hence the headroom.
294            anyhow::bail!(
295                "Long-context probe: {} tokens plus {LONG_CTX_HEADROOM} tokens of needle, \
296                 question and answer does not fit in a {max_seq_len}-token context",
297                self.long_ctx_tokens
298            );
299        }
300        self.opts = Some(ServeOptions {
301            max_seq_len,
302            speculative: values.bool("speculative")?,
303        });
304        self.tps_tokens = values.usize("tps_tokens")?;
305        self.probe_budget = values.usize("probe_budget")?;
306        self.timeout = Duration::from_secs(values.usize("request_timeout_s")? as u64);
307        self.update_baselines = values.bool("update_baselines")?;
308        self.plan = Plan::default();
309        self.planned_built = false;
310        self.cursor = 0;
311        self.results.clear();
312        Ok(())
313    }
314
315    async fn next(&mut self) -> Result<BenchmarkResult> {
316        let handle = self.handle()?.clone();
317        handle.check_cancelled()?;
318
319        if !self.planned_built {
320            self.build_plan()?;
321            let total = self.plan.planned_count() as u64;
322            let skipped: Vec<String> = self
323                .plan
324                .skipped()
325                .map(|(r, why)| format!("{} ({})", r.model, why.reason()))
326                .collect();
327            let mut frame = BenchmarkResult::running("plan", self.elapsed())
328                .with_progress(0, total.max(1))
329                .with_table(report::table(&self.plan, &self.results))
330                .log_line(LogLine::info(format!(
331                    "{total} checkpoint(s) planned · {} not runnable on this box · {} outside the \
332                     filter",
333                    self.plan.skipped().count(),
334                    self.plan.excluded_count()
335                )));
336            if !skipped.is_empty() {
337                frame = frame.log_line(LogLine::info(format!("skipping {}", skipped.join(", "))));
338            }
339            return Ok(frame);
340        }
341
342        let total = self.plan.planned_count();
343        if self.cursor >= total {
344            return self.finish();
345        }
346
347        // 2026-09-26: The same iterator `score::tally` walks.
348        let round = self
349            .plan
350            .planned()
351            .nth(self.cursor)
352            .cloned()
353            .context("the plan changed under the cursor")?;
354        let outcome = self.run_round(&round).await;
355        let result = RoundResult {
356            label: round.label(),
357            baseline_tps: self.baseline_for(&round.label()),
358            outcome,
359        };
360        let line = report::round_line(&result);
361        let failed = !result.bars().is_empty();
362        self.results.push(result);
363        self.cursor += 1;
364        let total = total as u64;
365        handle.progress(self.cursor as u64, total);
366        let tally = score::tally(&self.plan, &self.results);
367        Ok(BenchmarkResult::running(round.label(), self.elapsed())
368            .with_progress(self.cursor as u64, total)
369            .with_summary(report::summary(&tally))
370            .with_table(report::table(&self.plan, &self.results))
371            .log_line(if failed {
372                LogLine::error(line)
373            } else {
374                LogLine::info(line)
375            }))
376    }
377
378    /// 2026-09-26: Ask the host to put back the model that was serving. Before
379    /// a plan exists nothing was booted, so there is nothing to restore.
380    async fn cleanup(&mut self) -> Result<()> {
381        let Some(host) = self.host.clone() else {
382            return Ok(());
383        };
384        if !self.planned_built {
385            return Ok(());
386        }
387        host.restore().await.context("restoring the previous model")
388    }
389}
390
391#[cfg(test)]
392#[path = "serve_matrix_tests.rs"]
393mod tests;
metrale_bench/benchmarks/serve_matrix/mod.rs

metrale_bench/benchmarks/serve_matrix/
mod.rs
