use serde::{Deserialize, Serialize};
use super::bootstrap::{median_decode_tok_s, paired_ratio_lcb};
use super::join::{BandRatios, JoinKey, Ratio, RatioMethod};
use super::metrics::RequestSample;
use super::protocol::{stream_live_ttft_over_e2e_max, INTERLEAVED, REPLICATES};
use super::receipt::RunId;
use super::replicate::MIN_REPLICATES;
use super::samples::SamplesFile;
use super::witness::BatchInvarianceWitness;
pub const REQUEST_TIMEOUT_MS: f64 = 120_000.0;
pub const DRAIN_SUSPECT_FRACTION: f64 = 0.5;
pub const SCHEMA_VERSION: u32 = 3;
pub const VERDICT_CONFIDENCE: f64 = 0.95;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Outcome {
Completed,
Timeout,
AbandonedAtDrain,
Failed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StreamMode {
Live,
Replayed,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StreamVerdict {
Live,
Replayed,
Undeclared,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StreamWitnessSource {
Server,
Client,
}
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct StreamWitness {
pub client_ttft_over_e2e_median: f64,
pub verdict: StreamVerdict,
pub source: StreamWitnessSource,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum BandStatus {
Measured,
Unmeasured,
Na,
InvalidCorrectness,
NonconformantValid,
ComparatorStale,
}
impl BandStatus {
#[must_use]
pub fn wire_token(self) -> &'static str {
match self {
Self::Measured => "MEASURED",
Self::Unmeasured => "UNMEASURED",
Self::Na => "NA",
Self::InvalidCorrectness => "INVALID-CORRECTNESS",
Self::NonconformantValid => "NONCONFORMANT-VALID",
Self::ComparatorStale => "COMPARATOR_STALE",
}
}
#[must_use]
pub fn vocabulary() -> [Self; 6] {
[
Self::Measured,
Self::Unmeasured,
Self::Na,
Self::InvalidCorrectness,
Self::NonconformantValid,
Self::ComparatorStale,
]
}
#[must_use]
pub fn baseline_eligible(self) -> bool {
self == Self::Measured
}
#[must_use]
pub fn rank(self) -> u8 {
match self {
Self::InvalidCorrectness => 0,
Self::ComparatorStale => 1,
Self::Na => 2,
Self::NonconformantValid => 3,
Self::Unmeasured => 4,
Self::Measured => 5,
}
}
#[must_use]
pub fn stronger_of(self, other: Self) -> Self {
if other.rank() < self.rank() {
other
} else {
self
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Lane {
Apr,
Llama,
}
impl Lane {
#[must_use]
pub fn wire_token(self) -> &'static str {
match self {
Self::Apr => "apr",
Self::Llama => "llama",
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AdmissionCap {
pub lane: Lane,
pub cap: u32,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub enum ComparatorStatus {
NotApplicable {
decided_by: String,
reason: String,
budget: Option<String>,
},
Unmeasured {
owner: String,
reason: String,
admission_capped: Option<AdmissionCap>,
},
Measured(MeasuredJoin),
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MeasuredJoin {
baseline: Box<DerivedBand>,
ratios: BandRatios,
}
impl MeasuredJoin {
pub(crate) fn sealed(baseline: DerivedBand, ratios: BandRatios) -> Self {
Self {
baseline: Box::new(baseline),
ratios,
}
}
#[must_use]
pub fn baseline(&self) -> &DerivedBand {
&self.baseline
}
#[must_use]
pub fn ratios(&self) -> &BandRatios {
&self.ratios
}
}
impl ComparatorStatus {
#[must_use]
pub fn unmeasured(owner: impl Into<String>, reason: impl Into<String>) -> Self {
Self::Unmeasured {
owner: owner.into(),
reason: reason.into(),
admission_capped: None,
}
}
#[must_use]
pub fn not_applicable(decided_by: impl Into<String>, reason: impl Into<String>) -> Self {
Self::NotApplicable {
decided_by: decided_by.into(),
reason: reason.into(),
budget: None,
}
}
#[must_use]
pub fn wire_token(&self) -> &'static str {
match self {
Self::NotApplicable { .. } => "NOT_APPLICABLE",
Self::Unmeasured { .. } => "UNMEASURED",
Self::Measured(_) => "MEASURED",
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LaneConfig {
pub n_ctx_slot: Option<u32>,
pub kv_type: Option<String>,
pub fa: Option<bool>,
pub n_batch: Option<u32>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct RequestOutcome {
pub issued_ms: f64,
pub settled_ms: f64,
pub outcome: Outcome,
pub generated_tokens: u32,
pub prompt_tokens: u32,
pub expected_tokens: Option<u32>,
pub ttft_ms: Option<f64>,
pub prefill_ms: Option<f64>,
pub in_flight_at_start: u32,
pub token_times_ms: Vec<f64>,
}
impl RequestOutcome {
#[must_use]
pub fn new(issued_ms: f64, settled_ms: f64, outcome: Outcome, generated_tokens: u32) -> Self {
Self {
issued_ms,
settled_ms,
outcome,
generated_tokens,
prompt_tokens: 0,
expected_tokens: None,
ttft_ms: None,
prefill_ms: None,
in_flight_at_start: 0,
token_times_ms: Vec::new(),
}
}
#[must_use]
pub fn completed(issued_ms: f64, settled_ms: f64, generated_tokens: u32) -> Self {
Self::new(issued_ms, settled_ms, Outcome::Completed, generated_tokens)
}
#[must_use]
pub fn streamed(mut self, ttft_ms: f64, token_times_ms: Vec<f64>) -> Self {
self.ttft_ms = Some(ttft_ms);
self.token_times_ms = token_times_ms;
self
}
#[must_use]
pub fn server_prefill(mut self, prompt_tokens: u32, prefill_ms: f64) -> Self {
self.prompt_tokens = prompt_tokens;
self.prefill_ms = Some(prefill_ms);
self
}
#[must_use]
pub fn with_prompt_tokens(mut self, prompt_tokens: u32) -> Self {
self.prompt_tokens = prompt_tokens;
self
}
#[must_use]
pub fn expecting(mut self, expected_tokens: u32) -> Self {
self.expected_tokens = Some(expected_tokens);
self
}
#[must_use]
pub fn in_flight(mut self, in_flight_at_start: u32) -> Self {
self.in_flight_at_start = in_flight_at_start;
self
}
#[must_use]
pub fn duration_ms(&self) -> f64 {
self.settled_ms - self.issued_ms
}
#[must_use]
pub fn decode_tok_per_sec(&self) -> Option<f64> {
let (first, last) = (self.token_times_ms.first()?, self.token_times_ms.last()?);
let span_s = (last - first) / 1000.0;
let n = self.token_times_ms.len();
if n < 2 || span_s <= 0.0 {
return None;
}
Some((n as f64 - 1.0) / span_s)
}
#[must_use]
pub fn itl_gaps_ms(&self) -> Vec<f64> {
self.token_times_ms
.windows(2)
.map(|w| w[1] - w[0])
.collect()
}
#[must_use]
pub fn ttft_over_e2e(&self) -> Option<f64> {
let ttft = self.ttft_ms?;
let e2e = self.duration_ms();
if e2e <= 0.0 {
return None;
}
Some(ttft / e2e)
}
#[must_use]
pub fn to_sample(&self, index: usize, in_flight_fallback: u32) -> RequestSample {
RequestSample {
index,
worker: 0,
start_s: self.issued_ms / 1000.0,
end_s: self.settled_ms / 1000.0,
token_times_s: self.token_times_ms.iter().map(|t| t / 1000.0).collect(),
generated_tokens: self.generated_tokens,
prompt_tokens: self.prompt_tokens,
outcome: self.outcome,
in_flight_at_start: if self.in_flight_at_start == 0 {
in_flight_fallback as usize
} else {
self.in_flight_at_start as usize
},
drained: false,
}
}
#[must_use]
pub fn to_row(&self, index: usize) -> SampleRow {
SampleRow {
index,
issued_ms: self.issued_ms,
settled_ms: self.settled_ms,
outcome: self.outcome,
generated_tokens: self.generated_tokens,
prompt_tokens: self.prompt_tokens,
ttft_ms: self.ttft_ms,
in_flight_at_start: self.in_flight_at_start,
}
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct SampleRow {
pub index: usize,
pub issued_ms: f64,
pub settled_ms: f64,
pub outcome: Outcome,
pub generated_tokens: u32,
pub prompt_tokens: u32,
pub ttft_ms: Option<f64>,
pub in_flight_at_start: u32,
}
#[derive(Debug, Clone, Copy, PartialEq)]
pub struct BandContext {
pub schema_version: u32,
pub replicates: u32,
pub interleaved: bool,
pub comparator_stale: bool,
pub stream_live_ttft_over_e2e_max: f64,
}
impl Default for BandContext {
fn default() -> Self {
Self {
schema_version: SCHEMA_VERSION,
replicates: REPLICATES as u32,
interleaved: INTERLEAVED,
comparator_stale: false,
stream_live_ttft_over_e2e_max: stream_live_ttft_over_e2e_max(),
}
}
}
impl BandContext {
#[must_use]
pub fn at_schema_version(schema_version: u32) -> Self {
Self {
schema_version,
..Self::default()
}
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct BandInput {
pub concurrency: u32,
pub window_ms: f64,
pub replicate: u32,
pub requests: Vec<RequestOutcome>,
pub comparator: ComparatorStatus,
pub n_predict: Option<u32>,
pub stream_mode: Option<StreamMode>,
pub witness: Option<BatchInvarianceWitness>,
pub samples_file: Option<SamplesFile>,
pub lane: LaneConfig,
pub role: Lane,
pub conformance_violations: Vec<String>,
}
impl BandInput {
#[must_use]
pub fn new(
concurrency: u32,
window_ms: f64,
requests: Vec<RequestOutcome>,
comparator: ComparatorStatus,
) -> Self {
Self {
concurrency,
window_ms,
replicate: 1,
requests,
comparator,
n_predict: None,
stream_mode: None,
witness: None,
samples_file: None,
lane: LaneConfig::default(),
role: Lane::Apr,
conformance_violations: Vec::new(),
}
}
#[must_use]
pub fn role(mut self, role: Lane) -> Self {
self.role = role;
self
}
#[must_use]
pub fn conformance_violations(mut self, violations: Vec<String>) -> Self {
self.conformance_violations = violations;
self
}
#[must_use]
pub fn replicate(mut self, replicate: u32) -> Self {
self.replicate = replicate;
self
}
#[must_use]
pub fn n_predict(mut self, n_predict: u32) -> Self {
self.n_predict = Some(n_predict);
self
}
#[must_use]
pub fn stream_mode(mut self, stream_mode: StreamMode) -> Self {
self.stream_mode = Some(stream_mode);
self
}
#[must_use]
pub fn witness(mut self, witness: BatchInvarianceWitness) -> Self {
self.witness = Some(witness);
self
}
#[must_use]
pub fn samples_file(mut self, samples_file: SamplesFile) -> Self {
self.samples_file = Some(samples_file);
self
}
#[must_use]
pub fn lane(mut self, lane: LaneConfig) -> Self {
self.lane = lane;
self
}
pub fn derive(&self) -> Result<DerivedBand, String> {
self.derive_in(&BandContext::default())
}
pub fn derive_at(&self, schema_version: u32) -> Result<DerivedBand, String> {
self.derive_in(&BandContext::at_schema_version(schema_version))
}
pub fn derive_in(&self, ctx: &BandContext) -> Result<DerivedBand, String> {
self.validate()?;
let drain_ms = self.drain_ms();
let span_ms = self.span_ms();
let tokens_total = self.tokens_total();
let short_of_n_predict = self.short_of_n_predict();
let stream_witness = self.stream_witness(ctx.stream_live_ttft_over_e2e_max);
let stream_live = stream_witness.is_some_and(|w| w.verdict == StreamVerdict::Live);
let invalid_correctness = self.invalid_correctness(ctx);
let mut unproduced = Vec::new();
let latency = if stream_live {
Latency::from(self)
} else {
unproduced.push(self.stream_reason(stream_witness.as_ref()));
Latency::none()
};
let mut prefill = self.prefill_tok_per_sec();
if prefill.is_none() {
unproduced.push(format!(
"PP-4 c={}: prefill_tok_per_sec — no request carried a server-reported \
`timings.prompt_ms`, and a client-side prefill estimate is exactly the \
harness-inferred field PP-13 refuses",
self.concurrency
));
}
let mut aggregate = Some(rate_per_sec(tokens_total as f64, span_ms));
let mut latency = latency;
if invalid_correctness {
aggregate = None;
prefill = None;
latency.decode_tok_per_sec = None;
unproduced.push(format!(
"P-4 c={}: aggregate_tok_per_sec, decode_tok_per_sec and prefill_tok_per_sec — \
the band's batch-invariance witness (PP-26) is {} , so its throughput is not \
reported, not gated and never a baseline",
self.concurrency,
self.witness.as_ref().map_or_else(
|| "absent".to_string(),
|w| format!("{:?}", w.batch_invariance)
)
));
}
if short_of_n_predict > 0 {
unproduced.push(format!(
"PP-28 c={}: {short_of_n_predict} of {} completed requests did not reach \
n_predict — the sampler pin was not honoured, so this band is a record and not \
a baseline",
self.concurrency,
self.count(Outcome::Completed)
));
}
for violation in &self.conformance_violations {
unproduced.push(format!(
"§4.4.2 c={}: protocol violation observed by the driver — {violation}. The band \
is NONCONFORMANT-VALID: a record, cited, never a baseline.",
self.concurrency
));
}
let metrics_complete =
aggregate.is_some() && latency.decode_tok_per_sec.is_some() && prefill.is_some();
let status = self.status(
ctx,
invalid_correctness,
stream_live,
short_of_n_predict,
metrics_complete,
);
if status == BandStatus::NonconformantValid {
unproduced.push(format!(
"§7.4 c={}: this band is NONCONFORMANT-VALID — a historical record, cited, never \
a baseline",
self.concurrency
));
}
Ok(DerivedBand {
concurrency: self.concurrency,
replicate: self.replicate,
window_ms: self.window_ms,
drain_ms,
suspect: self.suspect(drain_ms),
requested: self.requests.len(),
completed: self.count(Outcome::Completed),
timeouts: self.count(Outcome::Timeout),
truncated: self.count(Outcome::AbandonedAtDrain),
errors: self.count(Outcome::Failed),
short_of_n_predict,
tokens_total,
span_ms,
aggregate_tok_per_sec: aggregate,
decode_tok_per_sec: latency.decode_tok_per_sec,
prefill_tok_per_sec: prefill,
ttft_p50_ms: latency.ttft_p50_ms,
ttft_p95_ms: latency.ttft_p95_ms,
itl_p50_ms: latency.itl_p50_ms,
itl_p95_ms: latency.itl_p95_ms,
latencies_ms: self.latencies_ms(),
samples: self.sample_rows(),
samples_file: self.samples_file.clone(),
stream_mode: self.stream_mode,
stream_witness,
witness: self.witness.clone(),
status,
join_key: None,
run_id: None,
unproduced,
comparator: self.comparator.clone(),
})
}
pub fn join_status(
subject: &Self,
comparator: &Self,
subject_key: &JoinKey,
comparator_key: &JoinKey,
run_ids: (&RunId, &RunId),
) -> Result<ComparatorStatus, String> {
Self::join_status_in(
subject,
comparator,
subject_key,
comparator_key,
run_ids,
&BandContext::default(),
)
}
pub fn join_status_in(
subject: &Self,
comparator: &Self,
subject_key: &JoinKey,
comparator_key: &JoinKey,
run_ids: (&RunId, &RunId),
ctx: &BandContext,
) -> Result<ComparatorStatus, String> {
let (subject_run, comparator_run) = run_ids;
if subject_run != comparator_run {
return Err(format!(
"PP-3: the comparator lane is run_id {} and the subject is run_id {} — a ratio is \
representable only against a baseline from the SAME run; two runs saw two \
thermal states, two free-VRAM figures and two schedulers",
comparator_run.as_str(),
subject_run.as_str()
));
}
subject_key.refuse_mismatch(comparator_key)?;
let subject_band = subject.derive_in(ctx)?;
let comparator_band = comparator.derive_in(ctx)?;
for (lane, band) in [("subject", &subject_band), ("comparator", &comparator_band)] {
if band.timeouts > 0 {
return Err(format!(
"PP-5: the {lane} lane at c={} recorded {} timeouts — a timed-out band cannot \
carry a ratio, because the requests that did not return are exactly the ones \
the ratio would have to account for",
band.concurrency, band.timeouts
));
}
}
let ratios = ratios_of(subject, comparator, &subject_band, &comparator_band)?;
Ok(ComparatorStatus::Measured(MeasuredJoin::sealed(
comparator_band
.with_run_id(comparator_run.clone())
.with_join_key(comparator_key.clone()),
ratios,
)))
}
pub fn join(
subject: &Self,
comparator: &Self,
subject_key: &JoinKey,
comparator_key: &JoinKey,
run_ids: (&RunId, &RunId),
) -> Result<DerivedBand, String> {
let status = Self::join_status(subject, comparator, subject_key, comparator_key, run_ids)?;
let joined = Self {
comparator: status,
..subject.clone()
};
Ok(joined
.derive()?
.with_run_id(run_ids.0.clone())
.with_join_key(subject_key.clone()))
}
fn completed_iter(&self) -> impl Iterator<Item = &RequestOutcome> {
self.requests
.iter()
.filter(|r| r.outcome == Outcome::Completed)
}
fn count(&self, outcome: Outcome) -> usize {
self.requests
.iter()
.filter(|r| r.outcome == outcome)
.count()
}
fn tokens_total(&self) -> u64 {
self.completed_iter()
.map(|r| u64::from(r.generated_tokens))
.sum()
}
fn short_of_n_predict(&self) -> usize {
self.completed_iter()
.filter(|r| {
r.expected_tokens
.or(self.n_predict)
.is_some_and(|want| r.generated_tokens != want)
})
.count()
}
fn stream_witness(&self, live_max: f64) -> Option<StreamWitness> {
let ratios: Vec<f64> = self
.completed_iter()
.filter_map(RequestOutcome::ttft_over_e2e)
.collect();
let median = percentile(&sorted(ratios), 0.50)?;
let client_live = median <= live_max;
let (verdict, source) = match (self.stream_mode, client_live) {
(Some(StreamMode::Replayed), _) => {
(StreamVerdict::Replayed, StreamWitnessSource::Server)
}
(Some(StreamMode::Live), true) => (StreamVerdict::Live, StreamWitnessSource::Server),
(Some(StreamMode::Live), false) => {
(StreamVerdict::Replayed, StreamWitnessSource::Client)
}
(None, true) => (StreamVerdict::Live, StreamWitnessSource::Client),
(None, false) => (StreamVerdict::Undeclared, StreamWitnessSource::Client),
};
Some(StreamWitness {
client_ttft_over_e2e_median: median,
verdict,
source,
})
}
fn stream_reason(&self, witness: Option<&StreamWitness>) -> String {
let verdict = witness.map_or(StreamVerdict::Undeclared, |w| w.verdict);
let observed = witness.map_or_else(
|| "no completed request reported a first-token instant".to_string(),
|w| format!("median(ttft/e2e)={:.3}", w.client_ttft_over_e2e_median),
);
format!(
"PP-27 c={}: decode_tok_per_sec, ttft_ms p50/p95 and itl_ms p50/p95 — stream verdict \
{verdict:?} ({observed}); a latency computed off a replayed or undeclared stream is a \
property of the replay, not of the server",
self.concurrency
)
}
fn invalid_correctness(&self, ctx: &BandContext) -> bool {
self.role == Lane::Apr
&& ctx.schema_version >= SCHEMA_VERSION
&& self.concurrency > 1
&& !self
.witness
.as_ref()
.is_some_and(BatchInvarianceWitness::passed)
}
fn status(
&self,
ctx: &BandContext,
invalid_correctness: bool,
stream_live: bool,
short_of_n_predict: usize,
metrics_complete: bool,
) -> BandStatus {
let v3 = ctx.schema_version >= SCHEMA_VERSION;
let nonconformant = self.count(Outcome::Timeout) > 0
|| !self.conformance_violations.is_empty()
|| (v3
&& (!ctx.interleaved
|| (ctx.replicates as usize) < MIN_REPLICATES
|| !stream_live
|| short_of_n_predict > 0
|| !metrics_complete));
let mut status = match self.comparator {
ComparatorStatus::Measured(_) => BandStatus::Measured,
ComparatorStatus::NotApplicable { .. } => BandStatus::Na,
ComparatorStatus::Unmeasured { .. } => BandStatus::Unmeasured,
};
if nonconformant {
status = status.stronger_of(BandStatus::NonconformantValid);
}
if ctx.comparator_stale {
status = status.stronger_of(BandStatus::ComparatorStale);
}
if invalid_correctness {
status = status.stronger_of(BandStatus::InvalidCorrectness);
}
status
}
fn drain_ms(&self) -> f64 {
let last = self
.requests
.iter()
.map(|r| r.settled_ms)
.fold(f64::NEG_INFINITY, f64::max);
(last - self.window_ms).max(0.0)
}
fn span_ms(&self) -> f64 {
let first = self
.requests
.iter()
.map(|r| r.issued_ms)
.fold(f64::INFINITY, f64::min);
let last = self
.completed_iter()
.map(|r| r.settled_ms)
.fold(f64::NEG_INFINITY, f64::max);
(last - first).max(0.0)
}
fn suspect(&self, drain_ms: f64) -> Vec<String> {
if self.window_ms > 0.0 && drain_ms > DRAIN_SUSPECT_FRACTION * self.window_ms {
return vec![format!(
"SUSPECT PP-10 c={}: drain_ms={drain_ms:.1} > 0.5 x window_ms={:.1} — one \
request dominated the window; re-run this band with a longer window",
self.concurrency, self.window_ms
)];
}
Vec::new()
}
fn latencies_ms(&self) -> Vec<f64> {
self.completed_iter()
.map(RequestOutcome::duration_ms)
.collect()
}
fn sample_rows(&self) -> Vec<SampleRow> {
self.requests
.iter()
.enumerate()
.map(|(i, r)| r.to_row(i))
.collect()
}
fn request_samples(&self) -> Vec<RequestSample> {
self.requests
.iter()
.enumerate()
.map(|(i, r)| r.to_sample(i, self.concurrency))
.collect()
}
fn decode_median(&self) -> Option<f64> {
let rates: Vec<f64> = self
.completed_iter()
.filter_map(RequestOutcome::decode_tok_per_sec)
.collect();
percentile(&sorted(rates), 0.50)
}
fn ttft_percentile(&self, p: f64) -> Option<f64> {
let v: Vec<f64> = self.completed_iter().filter_map(|r| r.ttft_ms).collect();
percentile(&sorted(v), p)
}
fn itl_percentile(&self, p: f64) -> Option<f64> {
let v: Vec<f64> = self
.completed_iter()
.flat_map(RequestOutcome::itl_gaps_ms)
.collect();
percentile(&sorted(v), p)
}
fn prefill_tok_per_sec(&self) -> Option<f64> {
let mut tokens = 0_u64;
let mut ms = 0.0_f64;
for r in self.completed_iter() {
if let Some(p) = r.prefill_ms {
if p > 0.0 {
tokens += u64::from(r.prompt_tokens);
ms += p;
}
}
}
if ms <= 0.0 || tokens == 0 {
return None;
}
Some(tokens as f64 / (ms / 1000.0))
}
fn validate(&self) -> Result<(), String> {
if self.requests.is_empty() {
return Err(format!(
"band c={}: no sampled requests — a band over zero requests is a vacuous pass, \
not a measurement",
self.concurrency
));
}
if self.window_ms.is_nan() || self.window_ms <= 0.0 {
return Err(format!(
"band c={}: window_ms={} — the window must have positive length or `drain_ms` \
and the SUSPECT fraction are both undefined",
self.concurrency, self.window_ms
));
}
for (i, r) in self.requests.iter().enumerate() {
validate_request(self.concurrency, i, r, self.window_ms)?;
}
Ok(())
}
}
struct Latency {
decode_tok_per_sec: Option<f64>,
ttft_p50_ms: Option<f64>,
ttft_p95_ms: Option<f64>,
itl_p50_ms: Option<f64>,
itl_p95_ms: Option<f64>,
}
impl Latency {
fn from(band: &BandInput) -> Self {
Self {
decode_tok_per_sec: band.decode_median(),
ttft_p50_ms: band.ttft_percentile(0.50),
ttft_p95_ms: band.ttft_percentile(0.95),
itl_p50_ms: band.itl_percentile(0.50),
itl_p95_ms: band.itl_percentile(0.95),
}
}
fn none() -> Self {
Self {
decode_tok_per_sec: None,
ttft_p50_ms: None,
ttft_p95_ms: None,
itl_p50_ms: None,
itl_p95_ms: None,
}
}
}
fn ratios_of(
subject: &BandInput,
comparator: &BandInput,
subject_band: &DerivedBand,
comparator_band: &DerivedBand,
) -> Result<BandRatios, String> {
let agg = window_ratio(
subject_band.aggregate_tok_per_sec,
comparator_band.aggregate_tok_per_sec,
)
.ok_or_else(|| {
format!(
"P-5: neither lane at c={} produced an aggregate throughput, so there is no agg ratio \
to form",
subject_band.concurrency
)
})?;
let dec = if subject_band.decode_tok_per_sec.is_some()
&& comparator_band.decode_tok_per_sec.is_some()
{
paired_ratio_lcb(
&subject.request_samples(),
&comparator.request_samples(),
median_decode_tok_s,
VERDICT_CONFIDENCE,
)
} else {
None
};
let prefill = window_ratio(
subject_band.prefill_tok_per_sec,
comparator_band.prefill_tok_per_sec,
);
Ok(BandRatios { agg, dec, prefill })
}
fn window_ratio(subject: Option<f64>, comparator: Option<f64>) -> Option<Ratio> {
let (s, c) = (subject?, comparator?);
if c <= 0.0 {
return None;
}
Some(Ratio::reporting_only(
s / c,
RatioMethod::ReplicateTLower,
1,
))
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct DerivedBand {
pub concurrency: u32,
pub replicate: u32,
pub window_ms: f64,
pub drain_ms: f64,
pub suspect: Vec<String>,
pub requested: usize,
pub completed: usize,
pub timeouts: usize,
pub truncated: usize,
pub errors: usize,
pub short_of_n_predict: usize,
pub tokens_total: u64,
pub span_ms: f64,
pub aggregate_tok_per_sec: Option<f64>,
pub decode_tok_per_sec: Option<f64>,
pub prefill_tok_per_sec: Option<f64>,
pub ttft_p50_ms: Option<f64>,
pub ttft_p95_ms: Option<f64>,
pub itl_p50_ms: Option<f64>,
pub itl_p95_ms: Option<f64>,
pub latencies_ms: Vec<f64>,
pub samples: Vec<SampleRow>,
pub samples_file: Option<SamplesFile>,
pub stream_mode: Option<StreamMode>,
pub stream_witness: Option<StreamWitness>,
pub witness: Option<BatchInvarianceWitness>,
pub status: BandStatus,
pub join_key: Option<JoinKey>,
pub run_id: Option<RunId>,
pub unproduced: Vec<String>,
pub comparator: ComparatorStatus,
}
impl DerivedBand {
#[must_use]
pub fn with_join_key(mut self, key: JoinKey) -> Self {
self.join_key = Some(key);
self
}
#[must_use]
pub fn with_run_id(mut self, run_id: RunId) -> Self {
self.run_id = Some(run_id);
self
}
#[must_use]
pub fn marked_comparator_stale(mut self, pin_expiry: &str, started_utc: &str) -> Self {
self.status = self.status.stronger_of(BandStatus::ComparatorStale);
self.unproduced.push(format!(
"PP-20 c={}: the comparator pin expired {pin_expiry}, before this run started \
{started_utc} — every ratio on this band is COMPARATOR_STALE and blocks MEASURED \
until the pin is refreshed",
self.concurrency
));
self
}
#[must_use]
pub fn baseline_eligible(&self) -> bool {
self.status.baseline_eligible()
}
}
fn validate_request(c: u32, i: usize, r: &RequestOutcome, window_ms: f64) -> Result<(), String> {
let at = format!("band c={c} request[{i}]");
if r.issued_ms >= window_ms {
return Err(format!(
"{at}: issued_ms={} >= T={window_ms} — PP-10: no request is issued at or after the \
window close, and its tokens are never counted",
r.issued_ms
));
}
if r.settled_ms < r.issued_ms {
return Err(format!(
"{at}: settled_ms={} precedes issued_ms={}",
r.settled_ms, r.issued_ms
));
}
validate_outcome(&at, r, window_ms)
}
fn validate_outcome(at: &str, r: &RequestOutcome, window_ms: f64) -> Result<(), String> {
let d = r.duration_ms();
match r.outcome {
Outcome::Completed if r.generated_tokens == 0 => Err(format!(
"{at}: completed with zero generated tokens — a zero-token response is a failure, not \
a fast request"
)),
Outcome::Timeout if d < REQUEST_TIMEOUT_MS => Err(format!(
"{at}: labelled Timeout but ran {d:.1} ms < the {REQUEST_TIMEOUT_MS} ms hard timeout \
(§3) — that is a Failed, and the two are separate counters"
)),
Outcome::Failed if d >= REQUEST_TIMEOUT_MS => Err(format!(
"{at}: labelled Failed but ran {d:.1} ms >= the {REQUEST_TIMEOUT_MS} ms hard timeout \
— that is a Timeout, which PP-5 makes fatal to this band's ratio"
)),
Outcome::AbandonedAtDrain if r.settled_ms < window_ms => Err(format!(
"{at}: labelled AbandonedAtDrain but settled at {}, before T={window_ms} — a request \
can only be abandoned during the drain",
r.settled_ms
)),
_ => Ok(()),
}
}
fn rate_per_sec(count: f64, span_ms: f64) -> f64 {
if span_ms <= 0.0 {
return 0.0;
}
count / (span_ms / 1000.0)
}
fn sorted(mut v: Vec<f64>) -> Vec<f64> {
v.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
v
}
#[must_use]
pub fn percentile(sorted_ascending: &[f64], p: f64) -> Option<f64> {
match sorted_ascending.len() {
0 => None,
1 => Some(sorted_ascending[0]),
n => {
let idx = (n as f64 - 1.0) * p;
let lo = idx.floor() as usize;
let hi = (lo + 1).min(n - 1);
let frac = idx - lo as f64;
Some(sorted_ascending[lo].mul_add(1.0 - frac, sorted_ascending[hi] * frac))
}
}
}
#[cfg(test)]
mod tests {
#![allow(non_snake_case)]
use super::*;
use crate::perf_gate::receipt::{TokenCountingMethod, Workload};
use crate::perf_gate::witness::BatchInvariance;
fn done(issued_ms: f64, dur_ms: f64, tokens: u32) -> RequestOutcome {
RequestOutcome::completed(issued_ms, issued_ms + dur_ms, tokens)
}
fn streamed(issued_ms: f64, dur_ms: f64, tokens: u32) -> RequestOutcome {
let ttft = dur_ms * 0.08;
let times: Vec<f64> = (0..tokens)
.map(|k| issued_ms + ttft + f64::from(k) * (dur_ms - ttft) / f64::from(tokens))
.collect();
done(issued_ms, dur_ms, tokens)
.streamed(ttft, times)
.server_prefill(512, dur_ms * 0.05)
}
fn unmeasured() -> ComparatorStatus {
ComparatorStatus::unmeasured("perf-gate", "no comparator lane on this cell yet (PP-25)")
}
fn band(window_ms: f64, requests: Vec<RequestOutcome>) -> BandInput {
BandInput::new(1, window_ms, requests, unmeasured())
}
fn passing_witness() -> BatchInvarianceWitness {
let tokens: Vec<u32> = (0..128).collect();
BatchInvarianceWitness::compare(&tokens, &tokens, 64).formed_at(4, "perf041")
}
fn conformant_band(concurrency: u32) -> BandInput {
let requests: Vec<RequestOutcome> = (0..8)
.map(|i| streamed(f64::from(i) * 100.0, 90.0 + f64::from(i), 128))
.collect();
BandInput::new(concurrency, 1000.0, requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live)
.witness(passing_witness())
}
#[test]
fn drain_ms_is_zero_when_nothing_straddles_the_window_close() {
let d = band(1000.0, vec![done(0.0, 100.0, 128), done(200.0, 100.0, 128)])
.derive()
.expect("valid band");
assert_eq!(d.drain_ms, 0.0);
assert!(d.suspect.is_empty(), "{:?}", d.suspect);
}
#[test]
fn drain_ms_varies_with_actual_drain_behaviour() {
let quiet = band(1000.0, vec![done(0.0, 100.0, 128), done(900.0, 50.0, 128)])
.derive()
.expect("valid band");
let straggler = band(1000.0, vec![done(0.0, 100.0, 128), done(900.0, 350.0, 128)])
.derive()
.expect("valid band");
assert_eq!(quiet.drain_ms, 0.0);
assert!((straggler.drain_ms - 250.0).abs() < 1e-9, "{straggler:?}");
assert_ne!(quiet.drain_ms, straggler.drain_ms);
}
#[test]
fn a_dominating_request_is_annotated_suspect() {
let d = band(1000.0, vec![done(0.0, 50.0, 128), done(900.0, 700.0, 128)])
.derive()
.expect("valid band");
assert!((d.drain_ms - 600.0).abs() < 1e-9, "{d:?}");
assert_eq!(d.suspect.len(), 1, "{:?}", d.suspect);
assert!(d.suspect[0].contains("drain_ms"), "{:?}", d.suspect);
}
#[test]
fn a_drain_just_under_half_the_window_is_not_suspect() {
let d = band(1000.0, vec![done(0.0, 50.0, 128), done(900.0, 599.0, 128)])
.derive()
.expect("valid band");
assert!((d.drain_ms - 499.0).abs() < 1e-9, "{d:?}");
assert!(d.suspect.is_empty(), "{:?}", d.suspect);
}
#[test]
fn a_request_issued_at_or_after_t_is_refused() {
let at_t = band(1000.0, vec![done(0.0, 10.0, 8), done(1000.0, 10.0, 8)]).derive();
let after_t = band(1000.0, vec![done(0.0, 10.0, 8), done(1500.0, 10.0, 8)]).derive();
for (label, got) in [("at T", at_t), ("after T", after_t)] {
let err = got.expect_err(label);
assert!(err.contains("PP-10"), "{label}: {err}");
}
}
#[test]
fn max_tokens_truncation_is_not_drain_truncation() {
let reqs: Vec<RequestOutcome> = (0..8)
.map(|i| done(f64::from(i) * 100.0, 90.0, 128))
.collect();
let d = band(1000.0, reqs).derive().expect("valid band");
assert_eq!(
d.truncated, 0,
"no request was abandoned at the drain deadline"
);
assert_eq!(d.completed, 8);
assert_eq!(d.tokens_total, 1024, "the numerator must not be emptied");
assert!(
d.aggregate_tok_per_sec.expect("agg") > 0.0,
"{:?}",
d.aggregate_tok_per_sec
);
}
#[test]
fn an_abandoned_request_increments_truncated_not_completed() {
let abandoned = RequestOutcome::new(900.0, 1400.0, Outcome::AbandonedAtDrain, 12);
let d = band(1000.0, vec![done(0.0, 100.0, 128), abandoned])
.derive()
.expect("valid band");
assert_eq!(d.truncated, 1);
assert_eq!(d.completed, 1);
assert_eq!(
d.tokens_total, 128,
"an abandoned request contributes no tokens"
);
assert!((d.drain_ms - 400.0).abs() < 1e-9, "{d:?}");
}
#[test]
fn an_abandonment_before_the_window_close_is_refused() {
let bogus = RequestOutcome::new(100.0, 200.0, Outcome::AbandonedAtDrain, 1);
let err = band(1000.0, vec![done(0.0, 10.0, 8), bogus])
.derive()
.expect_err("settled before T");
assert!(err.contains("only be abandoned during the drain"), "{err}");
}
#[test]
fn timeouts_and_failures_are_separate_and_both_are_verified() {
let timeout = RequestOutcome::new(10.0, 10.0 + REQUEST_TIMEOUT_MS, Outcome::Timeout, 0);
let failure = RequestOutcome::new(20.0, 45.0, Outcome::Failed, 0);
let with_both = band(
1000.0,
vec![done(0.0, 10.0, 8), timeout.clone(), failure.clone()],
);
let d = with_both.derive().expect("valid band");
assert_eq!(d.timeouts, 1);
assert_eq!(d.errors, 1);
assert_eq!(
d.requested,
d.completed + d.timeouts + d.truncated + d.errors,
"the four counters must partition the requests"
);
assert_eq!(
d.status,
BandStatus::NonconformantValid,
"PP-5: a band that timed out is a record, at every schema version"
);
assert_eq!(
with_both.derive_at(2).expect("renders").status,
BandStatus::NonconformantValid,
"…including a v2-dated one"
);
}
#[test]
fn a_short_request_cannot_be_labelled_a_timeout() {
let liar = RequestOutcome::new(0.0, 50.0, Outcome::Timeout, 0);
let err = band(1000.0, vec![done(500.0, 10.0, 8), liar])
.derive()
.expect_err("50 ms is not a timeout");
assert!(err.contains("hard timeout"), "{err}");
}
#[test]
fn an_over_long_request_cannot_be_labelled_a_plain_failure() {
let liar = RequestOutcome::new(0.0, REQUEST_TIMEOUT_MS + 1.0, Outcome::Failed, 0);
let err = band(1000.0, vec![done(500.0, 10.0, 8), liar])
.derive()
.expect_err("past the hard timeout");
assert!(err.contains("PP-5"), "{err}");
}
#[test]
fn a_zero_token_completion_is_refused() {
let err = band(1000.0, vec![done(0.0, 10.0, 0)])
.derive()
.expect_err("zero tokens");
assert!(err.contains("zero-token"), "{err}");
}
#[test]
fn an_empty_band_is_refused() {
let err = band(1000.0, Vec::new()).derive().expect_err("no requests");
assert!(err.contains("vacuous"), "{err}");
}
#[test]
fn a_non_positive_window_is_refused() {
let err = band(0.0, vec![done(-10.0, 5.0, 8)])
.derive()
.expect_err("zero window");
assert!(err.contains("window_ms"), "{err}");
}
#[test]
fn aggregate_is_wall_clock_over_the_whole_span() {
let d = band(
2500.0,
vec![done(0.0, 500.0, 100), done(1000.0, 1000.0, 100)],
)
.derive()
.expect("valid band");
assert!((d.span_ms - 2000.0).abs() < 1e-9, "{d:?}");
assert!(
(d.aggregate_tok_per_sec.expect("agg") - 100.0).abs() < 1e-9,
"200 tokens over 2 s = 100 tok/s, got {:?}",
d.aggregate_tok_per_sec
);
}
#[test]
fn a_non_streaming_band_names_what_it_could_not_produce() {
let d = band(1000.0, vec![done(0.0, 100.0, 128)])
.derive()
.expect("valid band");
assert_eq!(d.ttft_p50_ms, None);
assert_eq!(d.itl_p95_ms, None);
assert_eq!(d.decode_tok_per_sec, None);
assert_eq!(d.prefill_tok_per_sec, None);
assert_eq!(d.status, BandStatus::NonconformantValid);
let notes = d.unproduced.join("\n");
assert!(notes.contains("PP-27"), "{notes}");
assert!(notes.contains("PP-4"), "{notes}");
}
#[test]
fn a_streaming_band_produces_ttft_itl_and_decode() {
let one = RequestOutcome::completed(0.0, 500.0, 5)
.streamed(100.0, vec![100.0, 200.0, 300.0, 400.0, 500.0])
.server_prefill(512, 90.0);
let d = BandInput::new(1, 1000.0, vec![one], unmeasured())
.stream_mode(StreamMode::Live)
.n_predict(5)
.derive()
.expect("valid band");
assert_eq!(d.ttft_p50_ms, Some(100.0));
assert_eq!(d.itl_p50_ms, Some(100.0));
assert_eq!(d.decode_tok_per_sec, Some(10.0));
assert!(
(d.prefill_tok_per_sec.expect("prefill") - 512.0 / 0.09).abs() < 1e-6,
"{:?}",
d.prefill_tok_per_sec
);
assert!(d.unproduced.is_empty(), "{:?}", d.unproduced);
assert_eq!(d.status, BandStatus::Unmeasured, "no comparator lane");
}
#[test]
fn percentile_of_nothing_is_undefined_not_zero() {
assert_eq!(percentile(&[], 0.5), None);
assert_eq!(percentile(&[7.0], 0.95), Some(7.0));
assert_eq!(percentile(&[0.0, 10.0], 0.5), Some(5.0));
}
#[test]
fn comparator_status_renders_the_token_the_gate_reads() {
assert_eq!(unmeasured().wire_token(), "UNMEASURED");
let na = ComparatorStatus::not_applicable("perf-matrix.yaml", "vLLM has no aarch64 build");
assert_eq!(na.wire_token(), "NOT_APPLICABLE");
}
#[test]
fn status_tokens_are_exactly_the_section_7_4_vocabulary() {
let table = [
(BandStatus::Measured, "MEASURED"),
(BandStatus::Unmeasured, "UNMEASURED"),
(BandStatus::Na, "NA"),
(BandStatus::InvalidCorrectness, "INVALID-CORRECTNESS"),
(BandStatus::NonconformantValid, "NONCONFORMANT-VALID"),
(BandStatus::ComparatorStale, "COMPARATOR_STALE"),
];
assert_eq!(table.len(), BandStatus::vocabulary().len());
for (status, token) in table {
assert_eq!(status.wire_token(), token);
assert!(
BandStatus::vocabulary().contains(&status),
"{token} missing from the vocabulary"
);
}
assert_ne!(
BandStatus::Na.wire_token(),
"NOT_APPLICABLE",
"§7.4 spells it NA; NOT_APPLICABLE is the legacy comparator_status token"
);
assert!(BandStatus::Measured.baseline_eligible());
for s in BandStatus::vocabulary() {
if s != BandStatus::Measured {
assert!(!s.baseline_eligible(), "{s:?} may not be a baseline");
}
}
}
#[test]
fn a_completed_sample_short_of_n_predict_is_counted() {
let mut b = conformant_band(1);
b.requests[3].generated_tokens = 67;
let d = b.derive().expect("the band still renders");
assert_eq!(d.short_of_n_predict, 1);
assert_eq!(d.status, BandStatus::NonconformantValid);
assert!(
d.aggregate_tok_per_sec.is_some(),
"the evidence still renders; PP-28 is not fatal to the receipt"
);
let notes = d.unproduced.join("\n");
assert!(notes.contains("PP-28"), "{notes}");
}
#[test]
fn thirty_of_thirty_at_n_predict_pass() {
let requests: Vec<RequestOutcome> = (0..30)
.map(|i| streamed(f64::from(i) * 30.0, 90.0 + f64::from(i), 128))
.collect();
let d = BandInput::new(1, 1000.0, requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live)
.derive()
.expect("valid band");
assert_eq!(d.short_of_n_predict, 0);
assert_eq!(d.completed, 30);
assert_eq!(d.status, BandStatus::Unmeasured);
}
#[test]
fn a_band_with_short_samples_is_nonconformant() {
let mut b = conformant_band(4);
for r in &mut b.requests {
r.generated_tokens = 112;
}
let d = b.derive().expect("renders");
assert_eq!(d.short_of_n_predict, 8);
assert_eq!(d.status, BandStatus::NonconformantValid);
assert!(!d.baseline_eligible());
}
#[test]
fn a_per_request_expectation_overrides_the_band_pin() {
let mut b = conformant_band(1);
b.requests[0].generated_tokens = 64;
assert_eq!(b.derive().expect("renders").short_of_n_predict, 1);
b.requests[0] = b.requests[0].clone().expecting(64);
assert_eq!(b.derive().expect("renders").short_of_n_predict, 0);
}
#[test]
fn a_replayed_stream_sends_latency_to_unproduced() {
let d = conformant_band(1)
.stream_mode(StreamMode::Replayed)
.derive()
.expect("renders");
assert_eq!(d.decode_tok_per_sec, None);
assert_eq!(d.ttft_p95_ms, None);
assert_eq!(d.itl_p95_ms, None);
assert_eq!(
d.stream_witness.expect("witness").verdict,
StreamVerdict::Replayed
);
assert_eq!(d.status, BandStatus::NonconformantValid);
}
#[test]
fn a_server_claiming_live_is_overruled_by_the_client_witness() {
let late = RequestOutcome::completed(0.0, 500.0, 4)
.streamed(499.0, vec![499.0, 499.5, 499.8, 500.0])
.server_prefill(512, 40.0);
let d = BandInput::new(1, 1000.0, vec![late], unmeasured())
.stream_mode(StreamMode::Live)
.n_predict(4)
.derive()
.expect("renders");
let w = d.stream_witness.expect("witness");
assert!(w.client_ttft_over_e2e_median > 0.95, "{w:?}");
assert_eq!(w.verdict, StreamVerdict::Replayed);
assert_eq!(d.decode_tok_per_sec, None);
}
#[test]
fn the_stream_threshold_is_exclusive_at_the_declared_maximum() {
let ctx = BandContext {
stream_live_ttft_over_e2e_max: 0.95,
..BandContext::default()
};
let at_threshold = |ratio: f64| {
let e2e = 1000.0;
let ttft = ratio * e2e;
let one = RequestOutcome::completed(0.0, e2e, 4)
.streamed(ttft, vec![ttft, ttft + 10.0, ttft + 20.0, ttft + 30.0])
.server_prefill(512, 40.0);
BandInput::new(1, 2_000.0, vec![one], unmeasured())
.stream_mode(StreamMode::Live)
.n_predict(4)
.derive_in(&ctx)
.expect("renders")
};
assert_eq!(
at_threshold(0.95).stream_witness.expect("witness").verdict,
StreamVerdict::Live,
"exactly at the maximum is still live"
);
assert_eq!(
at_threshold(0.951).stream_witness.expect("witness").verdict,
StreamVerdict::Replayed
);
}
#[test]
fn an_undeclared_stream_the_client_measured_as_live_is_live() {
let d = BandInput::new(1, 1000.0, conformant_band(1).requests, unmeasured())
.n_predict(128)
.derive()
.expect("renders");
let w = d.stream_witness.expect("witness");
assert_eq!(w.verdict, StreamVerdict::Live);
assert_eq!(
w.source,
StreamWitnessSource::Client,
"the server said nothing"
);
assert_eq!(d.stream_mode, None, "and the receipt still says so");
assert!(d.decode_tok_per_sec.is_some(), "a live stream has a dec");
assert_eq!(
d.status,
BandStatus::Unmeasured,
"no comparator lane, but conformant"
);
}
#[test]
fn an_undeclared_stream_the_client_cannot_call_live_is_undeclared() {
let requests: Vec<RequestOutcome> = (0..6)
.map(|i| {
let issued = f64::from(i) * 10.0;
RequestOutcome::completed(issued, issued + 100.0 + f64::from(i), 128)
.streamed(99.0, vec![issued + 99.0, issued + 99.5, issued + 100.0])
})
.collect();
let d = BandInput::new(1, 1000.0, requests, unmeasured())
.n_predict(128)
.derive()
.expect("renders");
let w = d.stream_witness.expect("witness");
assert_eq!(w.verdict, StreamVerdict::Undeclared);
assert_eq!(w.source, StreamWitnessSource::Client);
assert_eq!(d.decode_tok_per_sec, None);
assert_eq!(d.ttft_p50_ms, None);
assert_eq!(d.itl_p95_ms, None);
assert_eq!(d.status, BandStatus::NonconformantValid);
}
#[test]
fn a_constant_token_batch_is_invalid_correctness() {
let m1: Vec<u32> = (0..128).map(|i| 1000 + i).collect();
let failing = BatchInvarianceWitness::compare(&m1, &vec![474_u32; 128], 64)
.formed_at(3, "scripts/perf041_batched_parity_probe.py");
let d = conformant_band(4)
.witness(failing)
.derive()
.expect("the band still renders");
assert_eq!(d.status, BandStatus::InvalidCorrectness);
assert_eq!(d.aggregate_tok_per_sec, None);
assert_eq!(d.decode_tok_per_sec, None);
assert_eq!(d.prefill_tok_per_sec, None);
}
#[test]
fn identical_128_token_prefixes_pass() {
let d = conformant_band(4).derive().expect("renders");
assert_eq!(
d.witness.expect("witness").batch_invariance,
BatchInvariance::Pass
);
assert_eq!(d.status, BandStatus::Unmeasured, "no comparator lane");
assert!(d.aggregate_tok_per_sec.is_some());
}
#[test]
fn an_invalid_correctness_band_reports_no_throughput() {
let d = conformant_band(8)
.witness(BatchInvarianceWitness::compare(&[1, 2, 3], &[9, 9, 9], 64))
.derive()
.expect("renders");
assert_eq!(d.status, BandStatus::InvalidCorrectness);
assert!(!d.baseline_eligible());
let notes = d.unproduced.join("\n");
assert!(notes.contains("aggregate_tok_per_sec"), "{notes}");
assert!(notes.contains("decode_tok_per_sec"), "{notes}");
assert!(notes.contains("prefill_tok_per_sec"), "{notes}");
}
#[test]
fn c1_needs_no_witness() {
let mut b = conformant_band(1);
b.witness = None;
let d = b.derive().expect("renders");
assert_ne!(d.status, BandStatus::InvalidCorrectness);
assert!(d.aggregate_tok_per_sec.is_some());
let mut wider = conformant_band(4);
wider.witness = None;
assert_eq!(
wider.derive().expect("renders").status,
BandStatus::InvalidCorrectness
);
}
#[test]
fn a_v2_receipt_is_historical_not_a_baseline() {
let mut b = conformant_band(4);
b.witness = None;
b.stream_mode = None;
let v2 = b.derive_at(2).expect("renders");
assert_ne!(v2.status, BandStatus::InvalidCorrectness);
assert!(
v2.aggregate_tok_per_sec.is_some(),
"a v2 band keeps its throughput"
);
assert!(!v2.baseline_eligible(), "but is never a baseline");
assert_eq!(
b.derive_at(3).expect("renders").status,
BandStatus::InvalidCorrectness,
"the same band at v3"
);
}
#[test]
fn a_measured_band_without_prefill_is_nonconformant() {
let mut b = conformant_band(1);
for r in &mut b.requests {
r.prefill_ms = None;
}
let d = b.derive().expect("renders");
assert_eq!(d.prefill_tok_per_sec, None);
assert_eq!(d.status, BandStatus::NonconformantValid);
assert!(d.unproduced.join("\n").contains("PP-13"));
}
#[test]
fn prefill_is_prompt_tokens_over_server_prefill_ms() {
let a = RequestOutcome::completed(0.0, 500.0, 8)
.streamed(
40.0,
vec![40.0, 100.0, 200.0, 300.0, 350.0, 400.0, 450.0, 500.0],
)
.server_prefill(500, 100.0);
let b = RequestOutcome::completed(10.0, 520.0, 8)
.streamed(
40.0,
vec![50.0, 110.0, 210.0, 310.0, 360.0, 410.0, 460.0, 520.0],
)
.server_prefill(300, 100.0);
let c = RequestOutcome::completed(20.0, 530.0, 8)
.streamed(
40.0,
vec![60.0, 120.0, 220.0, 320.0, 370.0, 420.0, 470.0, 530.0],
)
.with_prompt_tokens(9_999);
let zero = RequestOutcome::completed(30.0, 540.0, 8)
.streamed(
40.0,
vec![70.0, 130.0, 230.0, 330.0, 380.0, 430.0, 480.0, 540.0],
)
.server_prefill(7_777, 0.0);
let d = BandInput::new(1, 1000.0, vec![a, b, c, zero], unmeasured())
.stream_mode(StreamMode::Live)
.n_predict(8)
.derive()
.expect("renders");
assert!(
(d.prefill_tok_per_sec.expect("prefill") - 4_000.0).abs() < 1e-9,
"{:?}",
d.prefill_tok_per_sec
);
}
#[test]
fn fewer_than_five_replicates_makes_the_band_nonconformant() {
let b = conformant_band(1);
let five = BandContext {
replicates: 5,
..BandContext::default()
};
let three = BandContext {
replicates: 3,
..BandContext::default()
};
assert_eq!(
b.derive_in(&five).expect("renders").status,
BandStatus::Unmeasured
);
assert_eq!(
b.derive_in(&three).expect("renders").status,
BandStatus::NonconformantValid
);
}
#[test]
fn a_non_interleaved_receipt_makes_the_band_nonconformant() {
let ctx = BandContext {
interleaved: false,
..BandContext::default()
};
assert_eq!(
conformant_band(1).derive_in(&ctx).expect("renders").status,
BandStatus::NonconformantValid
);
}
#[test]
fn a_stale_pin_renders_comparator_stale() {
let ctx = BandContext {
comparator_stale: true,
..BandContext::default()
};
let d = conformant_band(1).derive_in(&ctx).expect("renders");
assert_eq!(d.status, BandStatus::ComparatorStale);
assert!(!d.baseline_eligible());
}
fn jkey(c: u32) -> JoinKey {
JoinKey {
host: "lambda".to_string(),
workload: Workload::W1,
band: c,
model: "qwen2.5-coder-7b-apache-q4k-v1".to_string(),
quant: "Q4_K_M".to_string(),
tokenization: TokenCountingMethod::ClientTokenizer,
window_ms: 1_000,
replicates: 5,
interleaved: true,
n_ctx_slot: Some(1024),
kv_type: Some("f16".to_string()),
fa: Some(true),
n_batch: Some(2048),
n_predict: 128,
}
}
fn same_run() -> RunId {
RunId::derive("2026-09-02T10:11:12.345Z", "lambda", &"a".repeat(64), 4242)
}
fn another_run() -> RunId {
RunId::derive("2026-09-02T11:00:00.000Z", "lambda", &"a".repeat(64), 4243)
}
#[test]
fn ratio_paired__a_same_run_baseline_joins() {
let subject = conformant_band(1);
let comparator = conformant_band(1);
let id = same_run();
let status = BandInput::join_status(&subject, &comparator, &jkey(1), &jkey(1), (&id, &id))
.expect("a same-run, same-key, timeout-free join");
let ComparatorStatus::Measured(join) = &status else {
panic!("expected Measured, got {status:?}");
};
let (baseline, ratios) = (join.baseline(), join.ratios());
assert_eq!(
baseline.run_id.as_ref(),
Some(&id),
"PP-3: the baseline says which run it came from"
);
assert_eq!(baseline.join_key.as_ref(), Some(&jkey(1)));
assert_eq!(status.wire_token(), "MEASURED");
assert!((ratios.agg.point - 1.0).abs() < 1e-9, "{:?}", ratios.agg);
assert_eq!(ratios.agg.method, RatioMethod::ReplicateTLower);
assert!(
ratios.agg.lcb95.is_none(),
"one replicate bounds no variance (§4.3)"
);
let dec = ratios.dec.as_ref().expect("a live stream has a dec ratio");
assert_eq!(dec.method, RatioMethod::PairedPercentileBootstrap);
assert!((dec.point - 1.0).abs() < 1e-9, "{dec:?}");
assert!(dec.lcb95.is_some(), "the request unit does bound");
assert!(ratios.prefill.is_some(), "both lanes reported prefill");
let joined =
BandInput::join(&subject, &comparator, &jkey(1), &jkey(1), (&id, &id)).expect("joins");
assert_eq!(joined.status, BandStatus::Measured);
assert!(joined.baseline_eligible());
}
#[test]
fn a_baseline_from_another_run_is_refused() {
let subject = conformant_band(1);
let comparator = conformant_band(1);
let (mine, theirs) = (same_run(), another_run());
assert_ne!(mine, theirs);
let err =
BandInput::join_status(&subject, &comparator, &jkey(1), &jkey(1), (&mine, &theirs))
.expect_err("cross-run baseline");
assert!(err.contains("PP-3"), "{err}");
assert!(err.contains("SAME run"), "{err}");
}
#[test]
fn a_key_mismatch_stops_the_join_before_any_ratio_is_computed() {
let id = same_run();
let err = BandInput::join_status(
&conformant_band(4),
&conformant_band(16),
&jkey(4),
&jkey(16),
(&id, &id),
)
.expect_err("c=4 against c=16");
assert!(err.contains("band: 4 != 16"), "{err}");
}
#[test]
fn a_timed_out_band_cannot_carry_a_ratio() {
let id = same_run();
let mut timed_out = conformant_band(1);
timed_out.requests.push(RequestOutcome::new(
10.0,
10.0 + REQUEST_TIMEOUT_MS,
Outcome::Timeout,
0,
));
assert_eq!(
timed_out.derive().expect("renders").timeouts,
1,
"control: the band itself still renders its evidence"
);
let subject_side = BandInput::join_status(
&timed_out,
&conformant_band(1),
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect_err("the subject timed out");
assert!(subject_side.contains("PP-5"), "{subject_side}");
assert!(subject_side.contains("subject"), "{subject_side}");
let comparator_side = BandInput::join_status(
&conformant_band(1),
&timed_out,
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect_err("the comparator timed out");
assert!(comparator_side.contains("comparator"), "{comparator_side}");
BandInput::join_status(
&conformant_band(1),
&conformant_band(1),
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect("a clean pair joins");
}
#[test]
fn the_joined_ratio_is_subject_over_comparator() {
let id = same_run();
let subject = conformant_band(1);
let mut fast_comparator = conformant_band(1);
for r in &mut fast_comparator.requests {
let dur = r.settled_ms - r.issued_ms;
r.settled_ms = r.issued_ms + dur / 2.0;
let first = r.token_times_ms[0];
for t in &mut r.token_times_ms {
*t = first + (*t - first) / 2.0;
}
}
let status =
BandInput::join_status(&subject, &fast_comparator, &jkey(1), &jkey(1), (&id, &id))
.expect("joins");
let ComparatorStatus::Measured(join) = &status else {
panic!("expected Measured");
};
let ratios = join.ratios();
assert!(
ratios.agg.point < 1.0,
"a slower subject is below parity: {:?}",
ratios.agg
);
let dec = ratios.dec.as_ref().expect("dec ratio");
assert!((dec.point - 0.5).abs() < 0.02, "{dec:?}");
}
#[test]
fn the_status_precedence_is_a_total_order_correctness_first() {
use BandStatus::{
ComparatorStale, InvalidCorrectness, Measured, Na, NonconformantValid, Unmeasured,
};
let strongest_first = [
InvalidCorrectness,
ComparatorStale,
Na,
NonconformantValid,
Unmeasured,
Measured,
];
for (i, strong) in strongest_first.iter().enumerate() {
for weak in &strongest_first[i + 1..] {
assert_eq!(
strong.stronger_of(*weak),
*strong,
"{strong:?} must win over {weak:?}"
);
assert_eq!(
weak.stronger_of(*strong),
*strong,
"…in either argument order"
);
}
assert_eq!(strong.stronger_of(*strong), *strong, "idempotent");
}
assert_eq!(strongest_first.len(), BandStatus::vocabulary().len());
}
#[test]
fn an_unwitnessed_batch_under_a_stale_pin_stays_invalid_correctness() {
let ctx = BandContext {
comparator_stale: true,
..BandContext::default()
};
let unwitnessed = BandInput::new(4, 1000.0, conformant_band(4).requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live);
let d = unwitnessed
.derive_in(&ctx)
.expect("renders")
.marked_comparator_stale("2026-01-01T00:00:00.000Z", "2026-09-02T10:11:12.345Z");
assert_eq!(d.status, BandStatus::InvalidCorrectness);
assert_eq!(
d.aggregate_tok_per_sec, None,
"and it reports no throughput"
);
assert!(!d.baseline_eligible());
let witnessed = conformant_band(4)
.derive_in(&ctx)
.expect("renders")
.marked_comparator_stale("2026-01-01T00:00:00.000Z", "2026-09-02T10:11:12.345Z");
assert_eq!(witnessed.status, BandStatus::ComparatorStale);
}
#[test]
fn a_not_applicable_band_is_na_even_when_it_is_also_nonconformant() {
let ctx = BandContext {
interleaved: false,
..BandContext::default()
};
let na = ComparatorStatus::not_applicable("perf-matrix.yaml", "no Metal path (#2841)");
let d = BandInput::new(1, 1000.0, conformant_band(1).requests, na)
.n_predict(128)
.stream_mode(StreamMode::Live)
.derive_in(&ctx)
.expect("renders");
assert_eq!(d.status, BandStatus::Na);
let d2 = conformant_band(1).derive_in(&ctx).expect("renders");
assert_eq!(d2.status, BandStatus::NonconformantValid);
}
#[test]
fn ratio_paired__the_measured_payload_is_read_only_outside_the_join() {
let id = same_run();
let status = BandInput::join_status(
&conformant_band(1),
&conformant_band(1),
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect("joins");
let ComparatorStatus::Measured(join) = &status else {
panic!("expected Measured");
};
assert_eq!(join.baseline().concurrency, 1);
assert_eq!(join.baseline().run_id.as_ref(), Some(&id));
assert!((join.ratios().agg.point - 1.0).abs() < 1e-9);
}
#[test]
fn a_comparator_lane_band_needs_no_batch_invariance_witness() {
let subject = BandInput::new(4, 1000.0, conformant_band(4).requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live);
let subject_band = subject.clone().derive().expect("renders");
assert_eq!(
subject_band.status,
BandStatus::InvalidCorrectness,
"the SUBJECT still needs one"
);
let comparator_band = subject.role(Lane::Llama).derive().expect("renders");
assert_ne!(comparator_band.status, BandStatus::InvalidCorrectness);
assert!(
comparator_band.aggregate_tok_per_sec.is_some(),
"the oracle's throughput is not withheld for a witness it is not the subject of"
);
assert!(
comparator_band.witness.is_none(),
"and it carries no witness of its own"
);
}
#[test]
fn the_comparator_exemption_is_about_the_lane_not_the_band_width() {
let one = BandInput::new(1, 1000.0, conformant_band(1).requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live);
assert_ne!(
one.clone().derive().expect("renders").status,
BandStatus::InvalidCorrectness
);
assert_ne!(
one.role(Lane::Llama).derive().expect("renders").status,
BandStatus::InvalidCorrectness
);
}
#[test]
fn a_lane_with_suppressed_decode_forms_no_dec_ratio() {
let id = same_run();
let replayed = BandInput::new(1, 1000.0, conformant_band(1).requests, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Replayed)
.witness(passing_witness());
assert_eq!(
replayed.derive().expect("renders").decode_tok_per_sec,
None,
"the fixture's decode must actually be withheld"
);
let status = BandInput::join_status(
&conformant_band(1),
&replayed,
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect("joins");
let ComparatorStatus::Measured(join) = &status else {
panic!("expected Measured");
};
assert!(
join.ratios().dec.is_none(),
"a ratio whose denominator the band refused to report is not a ratio: {:?}",
join.ratios().dec
);
let live = BandInput::join_status(
&conformant_band(1),
&conformant_band(1),
&jkey(1),
&jkey(1),
(&id, &id),
)
.expect("joins");
let ComparatorStatus::Measured(join) = &live else {
panic!("expected Measured");
};
assert!(join.ratios().dec.is_some());
}
#[test]
fn a_lane_without_server_prefill_forms_no_prefill_ratio() {
let id = same_run();
let no_timings: Vec<RequestOutcome> = (0..8)
.map(|i| {
let (issued, dur) = (f64::from(i) * 100.0, 90.0 + f64::from(i));
let ttft = dur * 0.08;
let times: Vec<f64> = (0..128)
.map(|k| issued + ttft + f64::from(k) * (dur - ttft) / 128.0)
.collect();
RequestOutcome::completed(issued, issued + dur, 128)
.with_prompt_tokens(512)
.streamed(ttft, times)
})
.collect();
let bare = BandInput::new(1, 1000.0, no_timings, unmeasured())
.n_predict(128)
.stream_mode(StreamMode::Live)
.witness(passing_witness());
assert_eq!(bare.derive().expect("renders").prefill_tok_per_sec, None);
let status =
BandInput::join_status(&conformant_band(1), &bare, &jkey(1), &jkey(1), (&id, &id))
.expect("joins");
let ComparatorStatus::Measured(join) = &status else {
panic!("expected Measured");
};
assert!(join.ratios().prefill.is_none());
}
#[test]
fn a_driver_protocol_violation_reaches_the_band_and_its_status() {
let clean = conformant_band(1).derive().expect("renders");
assert_eq!(clean.status, BandStatus::Unmeasured);
let violated = conformant_band(1)
.conformance_violations(vec![
"window closed after 30 samples, below the max(30, 8c) floor".to_string(),
])
.derive()
.expect("renders");
assert_eq!(violated.status, BandStatus::NonconformantValid);
assert!(
violated
.unproduced
.iter()
.any(|u| u.contains("below the max(30, 8c) floor") && u.contains("§4.4.2")),
"the violation text itself must be on the receipt: {:?}",
violated.unproduced
);
}
#[test]
fn a_band_carries_one_sample_row_per_request() {
let d = conformant_band(1).derive().expect("renders");
assert_eq!(d.samples.len(), d.requested);
assert_eq!(d.samples[0].index, 0);
assert_eq!(d.samples[0].generated_tokens, 128);
assert_eq!(d.samples[0].prompt_tokens, 512);
assert!(d.samples[0].ttft_ms.is_some());
let json = serde_json::to_string(&d.samples[0]).expect("serialises");
assert!(
!json.contains("token_times"),
"token times stay in the side file: {json}"
);
}
}