use std::time::Duration;
use serde::{Deserialize, Serialize};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ClientModel {
ClosedLoop,
}
pub const WARMUP_MULTIPLIER: usize = 2;
pub const QUIESCE: Duration = Duration::from_secs(5);
pub const MIN_SAMPLES_FLOOR: usize = 30;
pub const MIN_SAMPLES_PER_WORKER: usize = 8;
pub const MIN_WALL_CLOCK: Duration = Duration::from_secs(60);
pub const REQUEST_TIMEOUT: Duration = Duration::from_secs(120);
pub const COOLDOWN: Duration = Duration::from_secs(10);
pub const REPLICATES: usize = 5;
pub const INTERLEAVED: bool = true;
pub const N_PREDICT: u32 = 128;
pub const SAMPLER_TEMPERATURE: f64 = 0.0;
pub const SAMPLER_SEED: u64 = 0;
pub const SAMPLER_IGNORE_EOS: bool = true;
pub const STREAM_LIVE_TTFT_OVER_E2E_MAX: f64 = 0.95;
pub const WITNESS_MIN_AGREE_TOKENS: u32 = 64;
pub const BOOTSTRAP_RESAMPLES: usize = 10_000;
pub const BOOTSTRAP_SEED: u64 = 2026;
#[must_use]
pub fn min_sampled_requests(concurrency: usize) -> usize {
MIN_SAMPLES_FLOOR.max(MIN_SAMPLES_PER_WORKER * concurrency)
}
#[must_use]
pub fn warmup_requests(concurrency: usize) -> usize {
WARMUP_MULTIPLIER * concurrency
}
pub const PERF_MATRIX_SOURCE: &str = include_str!(concat!(env!("OUT_DIR"), "/perf-matrix.yaml"));
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Sampler {
pub temperature: f64,
pub seed: u64,
pub ignore_eos: bool,
}
impl Sampler {
#[must_use]
pub const fn spec_fallback() -> Self {
Self {
temperature: SAMPLER_TEMPERATURE,
seed: SAMPLER_SEED,
ignore_eos: SAMPLER_IGNORE_EOS,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ProtocolParams {
pub window_ms: u64,
pub warmup_requests_per_worker: u32,
pub quiesce_ms: u64,
pub cooldown_ms: u64,
pub n_predict: u32,
pub replicates: u32,
pub interleaved: bool,
pub sampler: Sampler,
}
impl ProtocolParams {
#[must_use]
pub const fn spec_fallback() -> Self {
Self {
window_ms: MIN_WALL_CLOCK.as_millis() as u64,
warmup_requests_per_worker: WARMUP_MULTIPLIER as u32,
quiesce_ms: QUIESCE.as_millis() as u64,
cooldown_ms: COOLDOWN.as_millis() as u64,
n_predict: N_PREDICT,
replicates: REPLICATES as u32,
interleaved: INTERLEAVED,
sampler: Sampler::spec_fallback(),
}
}
pub fn from_matrix() -> Result<Self, String> {
Self::from_matrix_source(PERF_MATRIX_SOURCE)
}
pub fn from_matrix_source(source: &str) -> Result<Self, String> {
let block = matrix_block(source, "protocol")?;
let raw: MatrixProtocolBlock = serde_yaml_ng::from_value(block)
.map_err(|e| format!("perf-matrix.yaml `protocol:` block: {e}"))?;
Ok(Self {
window_ms: raw.window_ms,
warmup_requests_per_worker: raw.warmup_requests_per_worker,
quiesce_ms: raw.quiesce_ms,
cooldown_ms: raw.cooldown_ms,
n_predict: raw.n_predict,
replicates: raw.replicates_min,
interleaved: raw.interleaved,
sampler: raw.sampler,
})
}
#[must_use]
pub fn effective_with_source() -> (Self, ProtocolSource) {
match Self::from_matrix() {
Ok(params) => (params, ProtocolSource::Matrix),
Err(reason) => (Self::spec_fallback(), ProtocolSource::SpecFallback(reason)),
}
}
#[must_use]
pub fn effective() -> Self {
Self::effective_with_source().0
}
pub fn source() -> Result<&'static str, String> {
Self::from_matrix().map(|_| "perf-matrix.yaml `protocol:`")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProtocolSource {
Matrix,
SpecFallback(String),
}
impl ProtocolSource {
#[must_use]
pub fn announcement(&self) -> String {
match self {
Self::Matrix => "protocol: matrix (scripts/perf-matrix.yaml `protocol:`)".to_string(),
Self::SpecFallback(reason) => {
format!("protocol: spec fallback because {reason}")
}
}
}
#[must_use]
pub fn unproduced_note(&self) -> Option<String> {
match self {
Self::Matrix => None,
Self::SpecFallback(reason) => Some(format!(
"PP-33 protocol — the `protocol:` block on this receipt is the compiled-in Rust \
spec fallback, NOT scripts/perf-matrix.yaml: {reason}. Every protocol parameter \
a gate compares against must live in the matrix; these came from consts and are \
unverifiable against it."
)),
}
}
}
pub fn stream_live_ttft_over_e2e_max_from(source: &str) -> Result<f64, String> {
let block = matrix_block(source, "stream")?;
let raw: MatrixStreamBlock = serde_yaml_ng::from_value(block)
.map_err(|e| format!("perf-matrix.yaml `stream:` block: {e}"))?;
Ok(raw.live_ttft_over_e2e_max)
}
#[must_use]
pub fn stream_live_ttft_over_e2e_max() -> f64 {
stream_live_ttft_over_e2e_max_from(PERF_MATRIX_SOURCE).unwrap_or(STREAM_LIVE_TTFT_OVER_E2E_MAX)
}
pub fn witness_min_agree_tokens_from(source: &str) -> Result<u32, String> {
let block = matrix_block(source, "witness")?;
let raw: MatrixWitnessBlock = serde_yaml_ng::from_value(block)
.map_err(|e| format!("perf-matrix.yaml `witness:` block: {e}"))?;
Ok(raw.min_agree_tokens)
}
#[must_use]
pub fn witness_min_agree_tokens() -> u32 {
witness_min_agree_tokens_from(PERF_MATRIX_SOURCE).unwrap_or(WITNESS_MIN_AGREE_TOKENS)
}
fn matrix_block(source: &str, key: &str) -> Result<serde_yaml_ng::Value, String> {
let doc: serde_yaml_ng::Value = serde_yaml_ng::from_str(source)
.map_err(|e| format!("perf-matrix.yaml does not parse as YAML: {e}"))?;
doc.get(key).cloned().ok_or_else(|| {
format!(
"perf-matrix.yaml has no `{key}:` block — PP-33 requires every protocol parameter and \
threshold to live there; refusing to substitute the Rust spec fallback silently"
)
})
}
#[derive(Debug, Deserialize)]
struct MatrixProtocolBlock {
window_ms: u64,
warmup_requests_per_worker: u32,
quiesce_ms: u64,
cooldown_ms: u64,
n_predict: u32,
replicates_min: u32,
interleaved: bool,
sampler: Sampler,
}
#[derive(Debug, Deserialize)]
struct MatrixStreamBlock {
live_ttft_over_e2e_max: f64,
}
#[derive(Debug, Deserialize)]
struct MatrixWitnessBlock {
min_agree_tokens: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct BandConfig {
pub concurrency: usize,
pub warmup_requests: usize,
pub quiesce: Duration,
pub min_samples: usize,
pub min_wall_clock: Duration,
pub request_timeout: Duration,
pub cooldown: Duration,
pub client_model: ClientModel,
}
impl BandConfig {
#[must_use]
pub fn conformant(concurrency: usize) -> Self {
let concurrency = concurrency.max(1);
Self {
concurrency,
warmup_requests: warmup_requests(concurrency),
quiesce: QUIESCE,
min_samples: min_sampled_requests(concurrency),
min_wall_clock: MIN_WALL_CLOCK,
request_timeout: REQUEST_TIMEOUT,
cooldown: COOLDOWN,
client_model: ClientModel::ClosedLoop,
}
}
#[must_use]
pub fn relaxed(
concurrency: usize,
min_samples: usize,
min_wall_clock: Duration,
quiesce: Duration,
) -> Self {
let concurrency = concurrency.max(1);
Self {
concurrency,
warmup_requests: warmup_requests(concurrency),
quiesce,
min_samples,
min_wall_clock,
request_timeout: REQUEST_TIMEOUT,
cooldown: COOLDOWN,
client_model: ClientModel::ClosedLoop,
}
}
#[must_use]
pub fn relaxed_with_cooldown(
concurrency: usize,
min_samples: usize,
min_wall_clock: Duration,
quiesce: Duration,
cooldown: Duration,
) -> Self {
Self {
cooldown,
..Self::relaxed(concurrency, min_samples, min_wall_clock, quiesce)
}
}
#[must_use]
pub fn conformance_violations(&self) -> Vec<String> {
self.conformance_violations_against(&ProtocolParams::effective())
}
#[must_use]
pub fn conformance_violations_against(&self, params: &ProtocolParams) -> Vec<String> {
let mut out = Vec::new();
let want_warmup = params.warmup_requests_per_worker as usize * self.concurrency;
if self.warmup_requests < want_warmup {
out.push(format!(
"§4.4.2 warmup_requests={} < {}*c={want_warmup}",
self.warmup_requests, params.warmup_requests_per_worker
));
}
let want_quiesce = Duration::from_millis(params.quiesce_ms);
if self.quiesce < want_quiesce {
out.push(format!(
"§4.4.2 quiesce={:?} < {want_quiesce:?}",
self.quiesce
));
}
let want_samples = min_sampled_requests(self.concurrency);
if self.min_samples < want_samples {
out.push(format!(
"§4.4.2 min_samples={} < max(30, 8*c)={want_samples}",
self.min_samples
));
}
let want_window = Duration::from_millis(params.window_ms);
if self.min_wall_clock < want_window {
out.push(format!(
"§5.1 min_wall_clock={:?} < window_ms={want_window:?}",
self.min_wall_clock
));
}
let want_cooldown = Duration::from_millis(params.cooldown_ms);
if self.cooldown < want_cooldown {
out.push(format!(
"§5.1 cooldown={:?} < cooldown_ms={want_cooldown:?}",
self.cooldown
));
}
if self.request_timeout != REQUEST_TIMEOUT {
out.push(format!(
"§4.4.3 request_timeout={:?} != 120s",
self.request_timeout
));
}
out
}
#[must_use]
pub fn is_conformant(&self) -> bool {
self.conformance_violations().is_empty()
}
}
pub use super::drain::{Outcome, DRAIN_SUSPECT_FRACTION};
pub use super::receipt::{TokenCountingMethod, TokenizationBlock};
#[cfg(test)]
mod tests {
use super::*;
const FIXTURE_MATRIX: &str = "\
schema_version: 2
protocol:
window_ms: 60000
warmup_requests_per_worker: 2
quiesce_ms: 5000
cooldown_ms: 10000
n_predict: 128
prompt_tokens: 512
replicates_min: 5
interleaved: true
sampler: {temperature: 0.0, seed: 0, ignore_eos: true}
threshold_class: policy
author: spec-owner
stream:
live_ttft_over_e2e_max: 0.95
threshold_class: policy
author: spec-owner
witness:
min_agree_tokens: 64
threshold_class: policy
author: spec-owner
";
#[test]
fn min_sampled_requests_is_max_30_or_8c() {
assert_eq!(min_sampled_requests(1), 30);
assert_eq!(min_sampled_requests(3), 30);
assert_eq!(min_sampled_requests(4), 32);
assert_eq!(min_sampled_requests(8), 64);
assert_eq!(min_sampled_requests(16), 128);
}
#[test]
fn warmup_is_two_per_worker() {
for c in [1_usize, 4, 8, 16] {
assert_eq!(warmup_requests(c), 2 * c);
}
}
#[test]
fn conformant_config_matches_the_spec_literals() {
for c in [1_usize, 4, 8, 16] {
let cfg = BandConfig::conformant(c);
assert_eq!(cfg.concurrency, c);
assert_eq!(cfg.warmup_requests, 2 * c);
assert_eq!(cfg.quiesce, Duration::from_secs(5));
assert_eq!(cfg.min_samples, 30.max(8 * c));
assert_eq!(cfg.min_wall_clock, Duration::from_secs(60));
assert_eq!(cfg.request_timeout, Duration::from_secs(120));
assert_eq!(cfg.cooldown, Duration::from_secs(10));
assert_eq!(cfg.client_model, ClientModel::ClosedLoop);
assert!(
cfg.is_conformant(),
"violations: {:?}",
cfg.conformance_violations()
);
}
}
#[test]
fn a_matrix_without_a_protocol_block_is_an_error_not_a_default() {
let err = ProtocolParams::from_matrix_source("schema_version: 2\nbands: [1, 4]\n")
.expect_err("no protocol block");
assert!(err.contains("`protocol:`"), "{err}");
assert!(err.contains("PP-33"), "{err}");
}
#[test]
fn the_protocol_block_is_read_from_the_matrix() {
let p = ProtocolParams::from_matrix_source(FIXTURE_MATRIX).expect("block parses");
assert_eq!(p.window_ms, 60_000);
assert_eq!(p.warmup_requests_per_worker, 2);
assert_eq!(p.quiesce_ms, 5_000);
assert_eq!(p.cooldown_ms, 10_000);
assert_eq!(p.n_predict, 128);
assert_eq!(
p.replicates, 5,
"matrix `replicates_min` is the receipt's n floor"
);
assert!(p.interleaved);
assert_eq!(p.sampler.temperature, 0.0);
assert_eq!(p.sampler.seed, 0);
assert!(p.sampler.ignore_eos);
}
#[test]
fn conformance_violations_read_the_loaded_params_not_the_consts() {
let cfg = BandConfig::conformant(4);
let spec = ProtocolParams::spec_fallback();
assert!(cfg.conformance_violations_against(&spec).is_empty());
let wider = ProtocolParams {
window_ms: 120_000,
..spec
};
let v = cfg.conformance_violations_against(&wider);
assert_eq!(v.len(), 1, "{v:?}");
assert!(v[0].contains("min_wall_clock"), "{v:?}");
}
#[test]
fn a_missing_cooldown_is_a_conformance_violation() {
let cfg = BandConfig::relaxed_with_cooldown(
4,
32,
Duration::from_secs(60),
Duration::from_secs(5),
Duration::ZERO,
);
let v = cfg.conformance_violations();
assert_eq!(v.len(), 1, "{v:?}");
assert!(v[0].contains("cooldown"), "{v:?}");
}
#[test]
fn the_effective_params_say_where_they_came_from() {
let effective = ProtocolParams::effective();
match ProtocolParams::source() {
Ok(where_from) => {
assert_eq!(where_from, "perf-matrix.yaml `protocol:`");
assert_eq!(effective, ProtocolParams::from_matrix().expect("block"));
}
Err(reason) => {
assert!(reason.contains("`protocol:`"), "{reason}");
assert_eq!(effective, ProtocolParams::spec_fallback());
}
}
}
#[test]
fn the_stream_and_witness_thresholds_are_read_from_the_matrix() {
assert_eq!(
stream_live_ttft_over_e2e_max_from(FIXTURE_MATRIX).expect("stream block"),
0.95
);
assert_eq!(
witness_min_agree_tokens_from(FIXTURE_MATRIX).expect("witness block"),
64
);
assert!(stream_live_ttft_over_e2e_max_from("bands: [1]\n").is_err());
assert!(witness_min_agree_tokens_from("bands: [1]\n").is_err());
}
#[test]
fn the_shipped_matrix_block_when_present_agrees_with_the_spec_fallback() {
match ProtocolParams::from_matrix() {
Ok(loaded) => assert_eq!(
loaded,
ProtocolParams::spec_fallback(),
"scripts/perf-matrix.yaml `protocol:` disagrees with protocol.rs's fallback"
),
Err(reason) => assert!(
reason.contains("`protocol:`"),
"the only acceptable absence is a named one: {reason}"
),
}
}
#[test]
fn the_protocol_source_is_reported_and_the_fallback_says_why() {
let (params, source) = ProtocolParams::effective_with_source();
match ProtocolParams::from_matrix() {
Ok(from_matrix) => {
assert_eq!(source, ProtocolSource::Matrix);
assert_eq!(params, from_matrix);
assert!(source.announcement().contains("matrix"), "{source:?}");
assert!(
source.unproduced_note().is_none(),
"the matrix supplied them; nothing is unproduced"
);
}
Err(reason) => {
assert_eq!(source, ProtocolSource::SpecFallback(reason.clone()));
assert_eq!(params, ProtocolParams::spec_fallback());
}
}
let fallback =
ProtocolSource::SpecFallback("perf-matrix.yaml has no `protocol:` block".to_string());
assert_eq!(
fallback.announcement(),
"protocol: spec fallback because perf-matrix.yaml has no `protocol:` block"
);
let note = fallback
.unproduced_note()
.expect("a fallback is an unproduced field");
assert!(note.contains("PP-33"), "{note}");
assert!(note.contains("no `protocol:` block"), "{note}");
assert!(
note.contains("NOT scripts/perf-matrix.yaml"),
"the note must say the block on the wire is not the matrix's: {note}"
);
}
#[test]
fn relaxed_config_reports_every_departure() {
let cfg = BandConfig::relaxed(4, 8, Duration::from_millis(50), Duration::ZERO);
assert!(!cfg.is_conformant());
let v = cfg.conformance_violations();
assert_eq!(
v.len(),
3,
"expected quiesce+min_samples+min_wall, got {v:?}"
);
assert!(v.iter().any(|s| s.contains("quiesce")));
assert!(v.iter().any(|s| s.contains("min_samples")));
assert!(v.iter().any(|s| s.contains("min_wall_clock")));
}
#[test]
fn client_model_serializes_as_closed_loop() {
let j =
serde_json::to_string(&ClientModel::ClosedLoop).expect("ClientModel must serialize");
assert_eq!(j, "\"closed_loop\"");
}
#[test]
fn the_two_request_timeout_spellings_agree() {
assert_eq!(
REQUEST_TIMEOUT.as_millis(),
u128::from(super::super::drain::REQUEST_TIMEOUT_MS as u64)
);
}
#[test]
fn declared_method_and_available_counter_must_agree() {
let ct = TokenizationBlock::ClientTokenizer {
tokenizer_sha256: "c".repeat(64),
counts_special_tokens: true,
counts_prompt_echo: false,
};
assert!(ct.validate().is_ok());
assert!(ct.require_counter(false).is_err());
assert!(ct.require_counter(true).is_ok());
let su = TokenizationBlock::ServerUsage {
counts_special_tokens: true,
counts_prompt_echo: false,
};
assert!(su.require_counter(true).is_err());
assert!(su.require_counter(false).is_ok());
}
}