From c1462fe64df063da6944cbd4a0d1a0d6214d2d63 Mon Sep 17 00:00:00 2001 From: plotarmordev <299844489+plotarmordev@users.noreply.github.com> Date: Thu, 10 Sep 2026 16:53:02 +0800 Subject: [PATCH] feat(perf): evaluate predeclared observed-performance policies --- crates/grill-perf/src/evidence.rs | 119 +++- crates/grill-perf/src/lifecycle.rs | 22 +- crates/grill-perf/src/main.rs | 47 +- crates/grill-perf/src/model.rs | 4 +- crates/grill-perf/src/policy.rs | 599 ++++++++++++++++ crates/grill-perf/src/run.rs | 15 + crates/grill-perf/tests/cli.rs | 44 +- crates/grill-perf/tests/support/policy.rs | 815 ++++++++++++++++++++++ docs/performance/CONTRACT.md | 23 + docs/performance/README.md | 113 +++ 10 files changed, 1763 insertions(+), 38 deletions(-) create mode 100644 crates/grill-perf/src/policy.rs create mode 100644 crates/grill-perf/tests/support/policy.rs diff --git a/crates/grill-perf/src/evidence.rs b/crates/grill-perf/src/evidence.rs index d2b3cf4..9df4af9 100644 --- a/crates/grill-perf/src/evidence.rs +++ b/crates/grill-perf/src/evidence.rs @@ -6,7 +6,7 @@ use std::io::{Read, Write}; use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt}; use std::path::{Path, PathBuf}; -fn hex(bytes: &[u8]) -> String { +pub(crate) fn hex(bytes: &[u8]) -> String { const DIGITS: &[u8] = b"0123456789abcdef"; let mut output = String::with_capacity(bytes.len() * 2); for byte in bytes { @@ -127,7 +127,7 @@ pub fn throughput(attempts: &[Attempt], elapsed_us: u64) -> (bool, Option, .map(|n| n as f64 * 1_000_000.0 / elapsed_us as f64); (eligible && rate.is_some(), tokens, rate) } -fn decode_rate(attempt: &Attempt) -> Option { +pub(crate) fn decode_sample(attempt: &Attempt) -> Option<(u64, u64)> { if attempt.status != Status::Complete { return None; } @@ -137,9 +137,9 @@ fn decode_rate(attempt: &Attempt) -> Option { if tokens < 2 || settle <= first { return None; } - Some((tokens - 1) as f64 * 1_000_000.0 / (settle - first) as f64) + Some((tokens - 1, settle - first)) } -fn prefill_rate(attempt: &Attempt) -> Option { +pub(crate) fn prefill_sample(attempt: &Attempt) -> Option<(u64, u64)> { if attempt.status != Status::Complete || attempt.usage.cached_prompt_tokens.is_some_and(|n| n > 0) { @@ -150,7 +150,13 @@ fn prefill_rate(attempt: &Attempt) -> Option { if tokens == 0 || first == 0 { return None; } - Some(tokens as f64 * 1_000_000.0 / first as f64) + Some((tokens, first)) +} +fn decode_rate(attempt: &Attempt) -> Option { + decode_sample(attempt).map(|(n, d)| n as f64 * 1_000_000.0 / d as f64) +} +fn prefill_rate(attempt: &Attempt) -> Option { + prefill_sample(attempt).map(|(n, d)| n as f64 * 1_000_000.0 / d as f64) } pub struct Loaded { pub plan: Plan, @@ -158,8 +164,32 @@ pub struct Loaded { pub states: Vec<&'static str>, pub history: crate::lifecycle::History, pub metrics: Option, + pub policy: Option, + pub plan_sha256: String, + pub evidence_sha256: String, + pub lineage_sha256: String, } pub fn load(root: &Path) -> Result { + load_verified(root).map_err(|error| error.detail) +} +pub(crate) struct LoadError { + pub reason: crate::policy::Reason, + pub detail: String, +} +impl From for LoadError { + fn from(detail: String) -> Self { + Self { + reason: crate::policy::Reason::InvalidEvidence, + detail, + } + } +} +impl From<&str> for LoadError { + fn from(detail: &str) -> Self { + detail.to_owned().into() + } +} +pub(crate) fn load_verified(root: &Path) -> Result { directory(root)?; let plan_bytes = read(&root.join("plan.json"), 8 * 1024 * 1024)?; let plan: Plan = decode(&plan_bytes)?; @@ -212,8 +242,36 @@ pub fn load(root: &Path) -> Result { { return Err("workload or schedule identity mismatch".into()); } + let policy = plan + .policy_sha256 + .as_ref() + .map(|hash| { + let bytes = read(&root.join("policy.json"), crate::policy::CAP).map_err(|detail| { + LoadError { + reason: crate::policy::Reason::InvalidPolicy, + detail, + } + })?; + if digest(&bytes) != *hash { + return Err(LoadError { + reason: crate::policy::Reason::PolicyHashMismatch, + detail: "policy sidecar hash mismatch".into(), + }); + } + crate::policy::parse(&bytes, &plan).map_err(|reason| LoadError { + detail: reason.as_str().into(), + reason, + }) + }) + .transpose()?; crate::wire::endpoint(&plan.endpoint, plan.local_http)?; let plan_hash = digest(&plan_bytes); + let mut fingerprint = Sha256::new(); + fingerprint.update(b"grill-perf-evidence-v1\0"); + fingerprint.update(plan_hash.as_bytes()); + fingerprint.update(plan.source_sha256.as_bytes()); + let mut lineage = Sha256::new(); + lineage.update(b"grill-perf-acquisition-lineage-v1\0"); let mut waves = Vec::with_capacity(plan.waves.len()); let mut states = Vec::with_capacity(plan.waves.len()); let mut metrics_budget = crate::metrics::Budget::default(); @@ -221,6 +279,7 @@ pub fn load(root: &Path) -> Result { for spec in &plan.waves { let dir = wave_dir(root, spec.index); if !exists(&dir)? { + fingerprint.update(b"not_started\0"); waves.push(None); states.push("not_started"); continue; @@ -228,6 +287,8 @@ pub fn load(root: &Path) -> Result { directory(&dir)?; let reservation_bytes = read(&dir.join("reservation.json"), 40 * 1024 * 1024)?; let reservation: Reservation = decode(&reservation_bytes)?; + let reservation_hash = digest(&reservation_bytes); + fingerprint.update(reservation_hash.as_bytes()); if reservation.version != 1 || reservation.plan_sha256 != plan_hash || reservation.wave != *spec @@ -251,14 +312,33 @@ pub fn load(root: &Path) -> Result { } let path = dir.join("wave.json"); if !exists(&path)? { + fingerprint.update(b"reserved_unsettled\0"); waves.push(None); states.push("reserved_unsettled"); continue; } - let wave: Wave = decode(&read(&path, FILE_CAP)?)?; + let wave_bytes = read(&path, FILE_CAP)?; + fingerprint.update(b"published\0"); + fingerprint.update(digest(&wave_bytes).as_bytes()); + let wave: Wave = decode(&wave_bytes)?; + lineage.update( + digest( + &serde_json::to_vec(&( + &wave.spec, + &wave.attempts, + wave.elapsed_us, + wave.dispatch_spread_us, + wave.preparation_us, + wave.reservation_publication_us, + wave.body_publication_us, + )) + .map_err(|e| e.to_string())?, + ) + .as_bytes(), + ); if wave.version != 1 || wave.plan_sha256 != plan_hash - || wave.reservation_sha256 != digest(&reservation_bytes) + || wave.reservation_sha256 != reservation_hash || wave.spec != *spec || wave.attempts.len() != spec.concurrency as usize { @@ -363,6 +443,8 @@ pub fn load(root: &Path) -> Result { states.push("published"); } let history = crate::lifecycle::history(root, &plan, &plan_hash, &states, &waves)?; + fingerprint.update(history.evidence_sha256.as_bytes()); + lineage.update(history.lineage_sha256.as_bytes()); Ok(Loaded { metrics: plan.metrics.as_ref().map(|config| crate::metrics::Summary { config: config.clone(), @@ -373,6 +455,10 @@ pub fn load(root: &Path) -> Result { waves, states, history, + policy, + plan_sha256: plan_hash, + evidence_sha256: hex(&fingerprint.finalize()), + lineage_sha256: hex(&lineage.finalize()), }) } #[derive(Serialize)] @@ -646,7 +732,7 @@ pub struct ReferenceIdentity { pub reasons: Vec, pub scope: &'static str, } -fn reference_identity(a: &Plan, a2: &Plan) -> ReferenceIdentity { +pub(crate) fn reference_identity(a: &Plan, a2: &Plan) -> ReferenceIdentity { let mut reasons = Vec::new(); let mut mismatch = false; let mut declaration = |name: &str, a: Option<&str>, a2: Option<&str>| match ( @@ -718,6 +804,15 @@ fn complete_lane_observations(values: &[Vec>]) -> bool { values.iter().flatten().all(Option::is_some) } +pub(crate) fn compatible(a: &Plan, b: &Plan) -> bool { + a.workload_sha256 == b.workload_sha256 + && a.tool_version == b.tool_version + && a.collector_sha256 == b.collector_sha256 + && a.local_http == b.local_http + && a.pool_max_idle_per_host == b.pool_max_idle_per_host + && a.metrics == b.metrics +} + pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result { let left = load(a)?; let right = load(b)?; @@ -727,13 +822,7 @@ pub fn compare(a: &Path, b: &Path, reference: Option<&Path>) -> Result, pub open: bool, + pub evidence_sha256: String, + pub lineage_sha256: String, } pub fn dir(root: &Path, index: usize) -> PathBuf { @@ -115,6 +118,8 @@ pub fn history( next_wave: 0, last_status: None, open: false, + evidence_sha256: evidence::digest(b"legacy-session-absent"), + lineage_sha256: evidence::digest(b"legacy-session-absent"), }); } if indices.is_empty() { @@ -124,15 +129,20 @@ pub fn history( let mut last_status = None; let mut open = false; let mut publication_uncertain = false; + let mut fingerprint = Sha256::new(); + let mut lineage = Sha256::new(); for (expected, index) in indices.iter().copied().enumerate() { if index != expected || open { return Err("noncontiguous or overlapping execution sessions".into()); } let path = dir(root, index); evidence::directory(&path)?; + let session_bytes = evidence::read(&path.join("session.json"), FILE_CAP)?; + fingerprint.update(evidence::digest(&session_bytes).as_bytes()); let session: Session = - serde_json::from_slice(&evidence::read(&path.join("session.json"), FILE_CAP)?) - .map_err(|e| format!("invalid session: {e}"))?; + serde_json::from_slice(&session_bytes).map_err(|e| format!("invalid session: {e}"))?; + lineage.update(session.started_unix_ms.to_le_bytes()); + lineage.update((session.first_wave as u64).to_le_bytes()); if session.version != 1 || session.index != index || session.first_wave != next_wave @@ -146,11 +156,15 @@ pub fn history( } if exists(&path.join("run.json"))? { let end_bytes = evidence::read(&path.join("run.json"), FILE_CAP)?; + fingerprint.update(b"settled\0"); + fingerprint.update(evidence::digest(&end_bytes).as_bytes()); + lineage.update(evidence::digest(&end_bytes).as_bytes()); let end: Summary = serde_json::from_slice(&end_bytes) .map_err(|e| format!("invalid session outcome: {e}"))?; if index == 0 { let receipt = root.join("run.json"); if exists(&receipt)? { + fingerprint.update(b"root-present\0"); if evidence::read(&receipt, FILE_CAP)? != end_bytes { return Err("first root receipt differs from session zero outcome".into()); } @@ -158,6 +172,7 @@ pub fn history( // Session publication precedes root publication. Do not repair or // mistake this crash window (or later loss) for settled evidence. publication_uncertain = true; + fingerprint.update(b"root-missing\0"); } } if end.version != 1 @@ -217,6 +232,7 @@ pub fn history( next_wave = end_wave; last_status = Some(end.status); } else { + fingerprint.update(b"unsettled\0"); open = true; last_status = None; } @@ -238,6 +254,8 @@ pub fn history( next_wave, last_status, open, + evidence_sha256: evidence::hex(&fingerprint.finalize()), + lineage_sha256: evidence::hex(&lineage.finalize()), }) } diff --git a/crates/grill-perf/src/main.rs b/crates/grill-perf/src/main.rs index 208e9c9..68dafad 100644 --- a/crates/grill-perf/src/main.rs +++ b/crates/grill-perf/src/main.rs @@ -2,6 +2,7 @@ mod evidence; mod lifecycle; mod metrics; mod model; +mod policy; mod run; mod wire; @@ -40,6 +41,15 @@ enum Command { #[arg(long)] json: bool, }, + /// Apply the captured observed-envelope policy to verified offline evidence. + Decide { + baseline: PathBuf, + candidate: PathBuf, + #[arg(long)] + reference: Option, + #[arg(long)] + json: bool, + }, } fn print_json(value: &impl serde::Serialize) -> model::Result<()> { let encoded = serde_json::to_string_pretty(value).map_err(|e| e.to_string())?; @@ -57,17 +67,37 @@ fn print_json(value: &impl serde::Serialize) -> model::Result<()> { } writeln!(output).map_err(|e| e.to_string()) } -fn execute(cli: Cli) -> model::Result { +fn execute(cli: Cli) -> model::Result { match cli.command { - Command::Run(options) => show_summary(run::execute(&options)?, options.json), + Command::Run(options) => show_summary(run::execute(&options)?, options.json) + .map(|complete| if complete { 0 } else { 2 }), Command::Pause { run } => { lifecycle::pause(&run)?; println!( "pause requested; active wave will drain before admission stops; inspect session run.json for paused/completed outcome" ); - Ok(true) + Ok(0) + } + Command::Resume { run, json } => show_summary(run::resume(&run, json)?, json) + .map(|complete| if complete { 0 } else { 2 }), + Command::Decide { + baseline, + candidate, + reference, + json, + } => { + let decision = policy::decide(&baseline, &candidate, reference.as_deref()); + if json { + print_json(&decision)?; + } else { + // The same versioned envelope makes output completion observable. + println!( + "Observed policy decision; not a statistical or causal claim. Eligibility is separate from the policy outcome." + ); + print_json(&decision)?; + } + Ok(decision.decision.exit()) } - Command::Resume { run, json } => show_summary(run::resume(&run, json)?, json), Command::Compare { baseline, candidate, @@ -142,7 +172,11 @@ fn execute(cli: Cli) -> model::Result { } } } - Ok(comparison.changes.iter().all(|c| c.eligible)) + Ok(if comparison.changes.iter().all(|c| c.eligible) { + 0 + } else { + 2 + }) } } } @@ -181,8 +215,7 @@ fn main() -> std::process::ExitCode { } }; match execute(cli) { - Ok(true) => std::process::ExitCode::SUCCESS, - Ok(false) => std::process::ExitCode::from(2), + Ok(code) => std::process::ExitCode::from(code), Err(error) => { eprintln!("grill-perf: {}", error.escape_debug()); std::process::ExitCode::from(1) diff --git a/crates/grill-perf/src/model.rs b/crates/grill-perf/src/model.rs index b3730a5..fbbd224 100644 --- a/crates/grill-perf/src/model.rs +++ b/crates/grill-perf/src/model.rs @@ -107,7 +107,7 @@ pub struct Workload { pub cases: Vec, pub cells: Vec, } -fn identifier(s: &str) -> bool { +pub(crate) fn identifier(s: &str) -> bool { !s.is_empty() && s.len() <= 64 && s.bytes() @@ -363,6 +363,8 @@ pub struct Plan { pub waves: Vec, #[serde(skip_serializing_if = "Option::is_none")] pub metrics: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub policy_sha256: Option, } #[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)] #[serde(deny_unknown_fields)] diff --git a/crates/grill-perf/src/policy.rs b/crates/grill-perf/src/policy.rs new file mode 100644 index 0000000..cbc9e71 --- /dev/null +++ b/crates/grill-perf/src/policy.rs @@ -0,0 +1,599 @@ +use crate::{evidence, model::*}; +use serde::{Deserialize, Serialize}; +use std::collections::HashSet; +use std::path::Path; + +pub const CAP: usize = 64 * 1024; +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Policy { + version: u32, + method: String, + id: String, + collector_sha256: String, + workload_source_sha256: String, + min_trials: u32, + cells: Vec, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct CellPolicy { + cell: String, + metrics: Vec, +} +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct MetricPolicy { + metric: Metric, + max_regression_bps: u32, + max_reference_spread_bps: u32, +} +#[derive(Clone, Copy, Deserialize, Serialize, PartialEq, Eq, Hash)] +#[serde(rename_all = "snake_case")] +enum Metric { + WaveLatencyUs, + AchievedCompletionTokensPerSecond, + DecodeTokensPerSecond, + PrefillTokensPerSecond, +} +#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum Reason { + InvalidEvidence, + InvalidPolicy, + PolicyHashMismatch, + PolicySourceMismatch, + PolicyCollectorMismatch, + PolicyScopeMismatch, + InsufficientDeclaredTrials, + MissingPolicy, + ConflictingPolicy, + MissingReference, + IncompatibleEvidence, + ReferenceUnqualified, + RoleReuse, + DeclaredStartsOutOfOrder, + SessionIncomplete, + WarmupIncomplete, + MetricUnavailable, + OutputAmountsMismatch, + NonpositiveReference, + ReferenceSpreadExceeded, + EnvelopeStraddlesTolerance, + ArithmeticOverflow, + EvaluatorUnavailable, +} +impl Reason { + pub fn as_str(self) -> &'static str { + match self { + Self::InvalidEvidence => "invalid_evidence", + Self::InvalidPolicy => "invalid_policy", + Self::PolicyHashMismatch => "policy_hash_mismatch", + Self::PolicySourceMismatch => "policy_source_mismatch", + Self::PolicyCollectorMismatch => "policy_collector_mismatch", + Self::PolicyScopeMismatch => "policy_scope_mismatch", + Self::InsufficientDeclaredTrials => "insufficient_declared_trials", + Self::MissingPolicy => "missing_policy", + Self::ConflictingPolicy => "conflicting_policy", + Self::MissingReference => "missing_reference", + Self::IncompatibleEvidence => "incompatible_evidence", + Self::ReferenceUnqualified => "reference_unqualified", + Self::RoleReuse => "role_reuse", + Self::DeclaredStartsOutOfOrder => "declared_starts_out_of_order", + Self::SessionIncomplete => "session_incomplete", + Self::WarmupIncomplete => "warmup_incomplete", + Self::MetricUnavailable => "metric_unavailable", + Self::OutputAmountsMismatch => "output_amounts_mismatch", + Self::NonpositiveReference => "nonpositive_reference", + Self::ReferenceSpreadExceeded => "reference_spread_exceeded", + Self::EnvelopeStraddlesTolerance => "envelope_straddles_tolerance", + Self::ArithmeticOverflow => "arithmetic_overflow", + Self::EvaluatorUnavailable => "evaluator_unavailable", + } + } +} +fn sha(s: &str) -> bool { + s.len() == 64 + && s.bytes() + .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b)) +} +pub fn parse(bytes: &[u8], plan: &Plan) -> Result { + if bytes.len() > CAP { + return Err(Reason::InvalidPolicy); + } + let policy: Policy = serde_json::from_slice(bytes).map_err(|_| Reason::InvalidPolicy)?; + if policy.version != 1 + || policy.method != "observed-envelope-v1" + || !identifier(&policy.id) + || !sha(&policy.collector_sha256) + || !sha(&policy.workload_source_sha256) + || !(3..=100).contains(&policy.min_trials) + { + return Err(Reason::InvalidPolicy); + } + if policy.collector_sha256 != plan.collector_sha256 { + return Err(Reason::PolicyCollectorMismatch); + } + if policy.workload_source_sha256 != plan.source_sha256 { + return Err(Reason::PolicySourceMismatch); + } + if policy.cells.len() != plan.workload.cells.len() { + return Err(Reason::PolicyScopeMismatch); + } + let mut cells = HashSet::new(); + for declared in &policy.cells { + let cell = plan + .workload + .cells + .iter() + .find(|c| c.id == declared.cell) + .ok_or(Reason::PolicyScopeMismatch)?; + if !cells.insert(&declared.cell) + || declared.metrics.is_empty() + || declared.metrics.len() > 4 + { + return Err(Reason::PolicyScopeMismatch); + } + if cell.trials < policy.min_trials { + return Err(Reason::InsufficientDeclaredTrials); + } + let mut metrics = HashSet::new(); + for metric in &declared.metrics { + if !metrics.insert(metric.metric) + || metric.max_regression_bps > 9999 + || metric.max_reference_spread_bps > 1_000_000 + { + return Err(Reason::InvalidPolicy); + } + } + } + Ok(policy) +} + +// Declaration order is also the aggregate precedence, not a vote across gates. +#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq, PartialOrd, Ord)] +#[serde(rename_all = "UPPERCASE")] +pub enum Outcome { + Pass, + Inconclusive, + Regression, + Error, +} +impl Outcome { + pub fn exit(self) -> u8 { + match self { + Self::Pass => 0, + Self::Error => 1, + Self::Inconclusive => 2, + Self::Regression => 3, + } + } +} +#[derive(Clone, Copy, Serialize)] +struct Rational { + numerator: u64, + denominator: u64, +} +impl From<(u64, u64)> for Rational { + fn from((numerator, denominator): (u64, u64)) -> Self { + Self { + numerator, + denominator, + } + } +} +fn order(a: Rational, b: Rational) -> std::cmp::Ordering { + (u128::from(a.numerator) * u128::from(b.denominator)) + .cmp(&(u128::from(b.numerator) * u128::from(a.denominator))) +} +fn scaled(a: Rational, scale: u32, b: Rational) -> Result { + u128::from(a.numerator) + .checked_mul(u128::from(scale)) + .and_then(|n| n.checked_mul(u128::from(b.denominator))) + .ok_or(Reason::ArithmeticOverflow) +} +fn range(values: &[Rational]) -> Option<[Rational; 2]> { + let first = *values.first()?; + Some( + values + .iter() + .copied() + .fold([first, first], |[low, high], v| { + [ + if order(v, low).is_lt() { v } else { low }, + if order(v, high).is_gt() { v } else { high }, + ] + }), + ) +} +#[derive(Serialize)] +struct Coverage { + expected_waves: u32, + observed_waves: usize, + eligible_waves: usize, + expected_observations: u32, + observed_observations: usize, + expected_warmups: u32, + eligible_warmups: usize, +} +#[derive(Serialize)] +struct Roles { + baseline: T, + candidate: T, + reference: T, +} +#[derive(Serialize)] +struct RoleIdentity { + plan_sha256: String, + evidence_sha256: String, + collector_sha256: String, + workload_source_sha256: String, + policy_sha256: Option, +} +#[derive(Serialize)] +struct Gate { + cell: String, + metric: Metric, + max_regression_bps: u32, + max_reference_spread_bps: u32, + coverage: Roles, + ranges: Roles>, + pooled_reference_range: Option<[Rational; 2]>, + adverse_bounds: Option<[f64; 2]>, + decision: Outcome, + reason_codes: Vec, +} +#[derive(Serialize)] +pub struct Decision { + version: u32, + claim: &'static str, + pub decision: Outcome, + pub eligibility: bool, + policy_sha256: Option, + policy_id: Option, + min_trials: Option, + evaluator_sha256: Option, + roles: Roles>, + gates: Vec, + reason_codes: Vec, +} +fn observations( + run: Option<&evidence::Loaded>, + cell: &Cell, + metric: Metric, +) -> (Coverage, Vec) { + let mut coverage = Coverage { + expected_waves: cell.trials, + observed_waves: 0, + eligible_waves: 0, + expected_observations: cell.trials + * if matches!( + metric, + Metric::WaveLatencyUs | Metric::AchievedCompletionTokensPerSecond + ) { + 1 + } else { + cell.concurrency + }, + observed_observations: 0, + expected_warmups: cell.warmup_trials, + eligible_warmups: 0, + }; + let mut values = Vec::new(); + if let Some(run) = run { + for wave in run + .waves + .iter() + .flatten() + .filter(|w| w.spec.cell == cell.id) + { + if wave.spec.phase == Phase::Warmup { + coverage.eligible_warmups += usize::from(wave.eligible); + continue; + } + coverage.observed_waves += 1; + coverage.eligible_waves += usize::from(wave.eligible); + match metric { + Metric::WaveLatencyUs => { + if wave.elapsed_us > 0 { + values.push((wave.elapsed_us, 1).into()); + } + } + Metric::AchievedCompletionTokensPerSecond => { + if let Some(tokens) = wave + .completion_tokens + .filter(|_| wave.eligible && wave.elapsed_us > 0) + { + values.push((tokens, wave.elapsed_us).into()); + } + } + Metric::DecodeTokensPerSecond | Metric::PrefillTokensPerSecond => { + values.extend( + wave.attempts + .iter() + .filter_map(|a| { + if metric == Metric::DecodeTokensPerSecond { + evidence::decode_sample(a) + } else { + evidence::prefill_sample(a) + } + }) + .map(Rational::from), + ); + } + } + } + } + coverage.observed_observations = values.len(); + (coverage, values) +} +fn amounts_match(a: &evidence::Loaded, b: &evidence::Loaded, cell: &str) -> bool { + a.plan + .waves + .iter() + .enumerate() + .filter(|(_, s)| s.phase == Phase::Measured && s.cell == cell) + .all(|(i, _)| match (&a.waves[i], &b.waves[i]) { + (Some(a), Some(b)) => a.attempts.iter().zip(&b.attempts).all(|(a, b)| { + a.usage.completion_tokens.is_some() + && a.usage.completion_tokens == b.usage.completion_tokens + }), + _ => false, + }) +} +fn evaluate( + gate: &mut Gate, + pooled: [Rational; 2], + candidate: [Rational; 2], +) -> Result<(), Reason> { + let [low, high] = pooled; + let [c_low, c_high] = candidate; + gate.pooled_reference_range = Some(pooled); + if low.numerator == 0 { + return Err(Reason::NonpositiveReference); + } + if scaled(high, 10000, low)? > scaled(low, 10000 + gate.max_reference_spread_bps, high)? { + return Err(Reason::ReferenceSpreadExceeded); + } + let ratio = |a: Rational, b: Rational| { + (a.numerator as f64 / a.denominator as f64) / (b.numerator as f64 / b.denominator as f64) + }; + let latency = gate.metric == Metric::WaveLatencyUs; + gate.adverse_bounds = Some(if latency { + [ratio(c_low, high) - 1.0, ratio(c_high, low) - 1.0] + } else { + [1.0 - ratio(c_high, low), 1.0 - ratio(c_low, high)] + }); + let tolerance = gate.max_regression_bps; + let (pass, regression) = if latency { + ( + scaled(c_high, 10000, low)? <= scaled(low, 10000 + tolerance, c_high)?, + scaled(c_low, 10000, high)? > scaled(high, 10000 + tolerance, c_low)?, + ) + } else { + ( + scaled(c_low, 10000, high)? >= scaled(high, 10000 - tolerance, c_low)?, + scaled(c_high, 10000, low)? < scaled(low, 10000 - tolerance, c_high)?, + ) + }; + gate.decision = if pass { + Outcome::Pass + } else if regression { + Outcome::Regression + } else { + return Err(Reason::EnvelopeStraddlesTolerance); + }; + Ok(()) +} +fn add(reasons: &mut Vec, reason: Reason) { + if !reasons.contains(&reason) { + reasons.push(reason); + } +} +pub fn decide(a: &Path, b: &Path, reference: Option<&Path>) -> Decision { + let mut result = Decision { + version: 1, + claim: "observed-policy-decision-not-statistical-or-causal", + decision: Outcome::Pass, + eligibility: false, + policy_sha256: None, + policy_id: None, + min_trials: None, + evaluator_sha256: None, + roles: Roles { + baseline: None, + candidate: None, + reference: None, + }, + gates: Vec::new(), + reason_codes: Vec::new(), + }; + match evidence::binary_digest() { + Ok(hash) => result.evaluator_sha256 = Some(hash), + Err(_) => { + result.decision = Outcome::Error; + add(&mut result.reason_codes, Reason::EvaluatorUnavailable); + } + } + let mut load = |path: Option<&Path>| match path.map(evidence::load_verified) { + Some(Ok(run)) if sha(&run.plan.collector_sha256) => Some(run), + Some(Ok(_)) => { + result.decision = Outcome::Error; + add(&mut result.reason_codes, Reason::InvalidEvidence); + None + } + Some(Err(error)) => { + result.decision = Outcome::Error; + add(&mut result.reason_codes, error.reason); + None + } + None => { + add(&mut result.reason_codes, Reason::MissingReference); + None + } + }; + let left = load(Some(a)); + let right = load(Some(b)); + let repeat = load(reference); + let runs = [left.as_ref(), right.as_ref(), repeat.as_ref()]; + let identity = |run: Option<&evidence::Loaded>| { + run.map(|r| RoleIdentity { + plan_sha256: r.plan_sha256.clone(), + evidence_sha256: r.evidence_sha256.clone(), + collector_sha256: r.plan.collector_sha256.clone(), + workload_source_sha256: r.plan.source_sha256.clone(), + policy_sha256: r.plan.policy_sha256.clone(), + }) + }; + result.roles = Roles { + baseline: identity(runs[0]), + candidate: identity(runs[1]), + reference: identity(runs[2]), + }; + let mut bound = None; + for run in runs.into_iter().flatten() { + match run.plan.policy_sha256.as_ref() { + None => add(&mut result.reason_codes, Reason::MissingPolicy), + Some(hash) => { + if bound.is_some_and(|prior| prior != hash) { + result.decision = Outcome::Error; + add(&mut result.reason_codes, Reason::ConflictingPolicy); + } + bound = Some(hash); + } + } + if run.history.count != 1 + || run.history.open + || run.history.last_status.as_deref() != Some("completed") + { + add(&mut result.reason_codes, Reason::SessionIncomplete); + } + if let Some(left) = &left + && !evidence::compatible(&left.plan, &run.plan) + { + result.decision = Outcome::Error; + add(&mut result.reason_codes, Reason::IncompatibleEvidence); + } + } + for i in 0..runs.len() { + for j in i + 1..runs.len() { + if let (Some(a), Some(b)) = (runs[i], runs[j]) { + if a.plan_sha256 == b.plan_sha256 || a.lineage_sha256 == b.lineage_sha256 { + add(&mut result.reason_codes, Reason::RoleReuse); + } + if a.plan.started_unix_ms >= b.plan.started_unix_ms { + add(&mut result.reason_codes, Reason::DeclaredStartsOutOfOrder); + } + } + } + } + if let (Some(a), Some(a2)) = (runs[0], runs[2]) + && evidence::reference_identity(&a.plan, &a2.plan).status + != evidence::ReferenceIdentityStatus::DeclaredMatch + { + add(&mut result.reason_codes, Reason::ReferenceUnqualified); + } + let qualified = result.reason_codes.is_empty(); + let invalid = result.decision == Outcome::Error; + if !qualified { + result.decision = result.decision.max(Outcome::Inconclusive); + } + if let Some(a) = &left { + result.policy_sha256 = a.plan.policy_sha256.clone(); + if let Some(policy) = &a.policy { + result.policy_id = Some(policy.id.clone()); + result.min_trials = Some(policy.min_trials); + for declared in &policy.cells { + // Policy admission already proved exact cell coverage. + let cell = a + .plan + .workload + .cells + .iter() + .find(|c| c.id == declared.cell) + .unwrap(); + for metric in &declared.metrics { + let [(ac, av), (bc, bv), (rc, rv)] = + runs.map(|r| observations(r, cell, metric.metric)); + let mut gate = Gate { + cell: cell.id.clone(), + metric: metric.metric, + max_regression_bps: metric.max_regression_bps, + max_reference_spread_bps: metric.max_reference_spread_bps, + coverage: Roles { + baseline: ac, + candidate: bc, + reference: rc, + }, + ranges: Roles { + baseline: range(&av), + candidate: range(&bv), + reference: range(&rv), + }, + pooled_reference_range: None, + adverse_bounds: None, + decision: Outcome::Inconclusive, + reason_codes: result.reason_codes.clone(), + }; + for coverage in [ + &gate.coverage.baseline, + &gate.coverage.candidate, + &gate.coverage.reference, + ] { + if coverage.expected_warmups == 0 + || coverage.eligible_warmups != coverage.expected_warmups as usize + { + add(&mut gate.reason_codes, Reason::WarmupIncomplete); + } + if coverage.eligible_waves != coverage.expected_waves as usize + || coverage.observed_observations + != coverage.expected_observations as usize + { + add(&mut gate.reason_codes, Reason::MetricUnavailable); + } + } + if let (Some(b), Some(r)) = (runs[1], runs[2]) + && evidence::compatible(&a.plan, &b.plan) + && evidence::compatible(&a.plan, &r.plan) + && (!amounts_match(a, b, &cell.id) || !amounts_match(a, r, &cell.id)) + { + add(&mut gate.reason_codes, Reason::OutputAmountsMismatch); + } + if gate.reason_codes.is_empty() + && let (Some(ar), Some(rr), Some(br)) = ( + gate.ranges.baseline, + gate.ranges.reference, + gate.ranges.candidate, + ) + { + let pooled = range(&[ar[0], ar[1], rr[0], rr[1]]).unwrap(); + if let Err(reason) = evaluate(&mut gate, pooled, br) { + gate.decision = if reason == Reason::ArithmeticOverflow { + Outcome::Error + } else { + Outcome::Inconclusive + }; + add(&mut gate.reason_codes, reason); + } + } else if invalid { + gate.decision = Outcome::Error; + } + result.decision = result.decision.max(gate.decision); + result.gates.push(gate); + } + } + } + } + result.eligibility = qualified + && result.gates.iter().all(|g| { + g.reason_codes.iter().all(|r| { + matches!( + r, + Reason::ReferenceSpreadExceeded + | Reason::EnvelopeStraddlesTolerance + | Reason::NonpositiveReference + ) + }) + }); + result +} diff --git a/crates/grill-perf/src/run.rs b/crates/grill-perf/src/run.rs index 10e70fa..a31ca32 100644 --- a/crates/grill-perf/src/run.rs +++ b/crates/grill-perf/src/run.rs @@ -43,6 +43,9 @@ pub struct Options { /// Optional JSON declarations of model revision, runtime, hardware and settings. #[arg(long)] pub deployment: Option, + /// Capture an exact observed-envelope declaration before dispatch. + #[arg(long)] + pub policy: Option, /// Optional bounded server-wide diagnostics; can perturb between-wave cadence. #[arg(long)] pub metrics_url: Option, @@ -118,6 +121,11 @@ pub fn execute(o: &Options) -> Result { }; let cache_mechanism = (workload.request.profile == Profile::VllmFixedV1) .then(|| "declared-vllm-prefix-cache".into()); + let policy_bytes = o + .policy + .as_ref() + .map(|path| evidence::read(path, crate::policy::CAP)) + .transpose()?; let plan = Plan { version: 2, kind: "performance-run-v1".into(), @@ -138,7 +146,11 @@ pub fn execute(o: &Options) -> Result { cache_namespace, waves, metrics, + policy_sha256: policy_bytes.as_deref().map(evidence::digest), }; + if let Some(bytes) = &policy_bytes { + crate::policy::parse(bytes, &plan).map_err(|e| e.as_str().to_owned())?; + } // Seed magnitude peaks at a corner of the trial/lane range and index 1023 is // the widest index, but lane 0 renders one digit narrower than lanes 10..63 in // each of the cache salt and the text salt, so an interior body can exceed a @@ -172,6 +184,9 @@ pub fn execute(o: &Options) -> Result { evidence::fresh(&o.out)?; let _owner = lifecycle::ownership(&o.out)?; evidence::write(&o.out.join("workload.json"), &source)?; + if let Some(bytes) = &policy_bytes { + evidence::write(&o.out.join("policy.json"), bytes)?; + } let plan_bytes = evidence::json(&o.out.join("plan.json"), &plan)?; evidence::sync(&o.out)?; let plan_hash = evidence::digest(&plan_bytes); diff --git a/crates/grill-perf/tests/cli.rs b/crates/grill-perf/tests/cli.rs index c79cef9..fc551a9 100644 --- a/crates/grill-perf/tests/cli.rs +++ b/crates/grill-perf/tests/cli.rs @@ -3050,16 +3050,17 @@ fn comparison_reference_pools_baseline_range_and_withholds_changes_inside_it() { #[test] fn comparison_change_outside_pooled_reference_range_remains_present() { let temp = Temp::new(); - let server = spread_server(&[ - (100, 100, 8), - (200, 200, 8), - (400, 400, 8), - (500, 500, 8), - (110, 110, 8), - (190, 190, 8), - ]); + let server = Server::new(|mut stream, _, _| { + header(&mut stream, "text/event-stream"); + frame(&mut stream, json!({"choices":[{"delta":{"content":"x"}}]})); + finish(&mut stream, Some(8), Some(0)); + }); let work = workload(1, 0, 2); - for name in ["a", "b", "a2"] { + for (name, first_times) in [ + ("a", [100_000u64, 200_000]), + ("b", [400_000, 500_000]), + ("a2", [50_000, 250_000]), + ] { successful(&run_declared( &temp, &server, @@ -3068,6 +3069,25 @@ fn comparison_change_outside_pooled_reference_range_remains_present() { "fixture-model", &deployment(), )); + // Synthetic observations make the pooled-envelope premise independent of + // scheduler load; the real CLI still verifies and compares the receipts. + for (index, first) in first_times.into_iter().enumerate() { + let path = temp.path(name).join(format!("wave-{index:06}/wave.json")); + let mut receipt = value(&path); + let elapsed = first * 2; + let timing = &mut receipt["attempts"][0]["timing"]; + timing["dispatch_offset_us"] = json!(0); + timing["headers_us"] = json!(1); + timing["first_body_us"] = json!(first); + timing["first_generated_text_us"] = json!(first); + timing["first_answer_text_us"] = json!(first); + timing["settle_us"] = json!(elapsed); + timing["capture_parse_us"] = json!(0); + receipt["elapsed_us"] = json!(elapsed); + receipt["dispatch_spread_us"] = json!(0); + receipt["achieved_completion_tokens_per_second"] = json!(8_000_000.0 / elapsed as f64); + fs::write(path, serde_json::to_vec(&receipt).unwrap()).unwrap(); + } } let output = cli() .arg("compare") @@ -3119,10 +3139,6 @@ fn comparison_change_outside_pooled_reference_range_remains_present() { let a = &report["baseline"][0]; let b = &report["candidate"][0]; let a2 = &report["reference"][0]; - assert!( - a[range][0].as_f64().unwrap() <= a2[range][1].as_f64().unwrap() - && a2[range][0].as_f64().unwrap() <= a[range][1].as_f64().unwrap() - ); let pooled_max = a[range][1] .as_f64() .unwrap() @@ -3772,3 +3788,5 @@ fn comparison_reference_lane_permutation_preserves_totals_but_is_ineligible() { } #[path = "support/metrics.rs"] mod metrics_tests; +#[path = "support/policy.rs"] +mod policy_tests; diff --git a/crates/grill-perf/tests/support/policy.rs b/crates/grill-perf/tests/support/policy.rs new file mode 100644 index 0000000..f3d7e0b --- /dev/null +++ b/crates/grill-perf/tests/support/policy.rs @@ -0,0 +1,815 @@ +use super::*; +use sha2::{Digest, Sha256}; + +const METRICS: [&str; 4] = [ + "wave_latency_us", + "achieved_completion_tokens_per_second", + "decode_tokens_per_second", + "prefill_tokens_per_second", +]; +fn hash(bytes: &[u8]) -> String { + Sha256::digest(bytes) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect() +} +fn save(path: impl AsRef, value: &Value) { + fs::write(path, serde_json::to_vec(value).unwrap()).unwrap(); +} +fn declaration(work: &Value, tolerance: u32, spread: u32) -> Value { + json!({"version":1,"method":"observed-envelope-v1","id":"synthetic-policy", + "collector_sha256":hash(&fs::read(env!("CARGO_BIN_EXE_grill-perf")).unwrap()), + "workload_source_sha256":hash(&serde_json::to_vec(work).unwrap()),"min_trials":3, + "cells":work["cells"].as_array().unwrap().iter().map(|cell| json!({"cell":cell["id"], + "metrics":METRICS.iter().map(|metric| json!({"metric":metric,"max_regression_bps":tolerance,"max_reference_spread_bps":spread})).collect::>() })).collect::>()}) +} +fn collect(temp: &Temp, server: &Server, name: &str, policy: bool) -> Output { + let mut command = cli(); + command + .arg("run") + .arg(temp.path("work.json")) + .args([ + "--endpoint", + &server.endpoint, + "--model", + "fixture-model", + "--local-http", + "--json", + ]) + .arg("--deployment") + .arg(temp.path("deployment.json")) + .arg("--out") + .arg(temp.path(name)); + if policy { + command.arg("--policy").arg(temp.path("policy.json")); + } + command.output().unwrap() +} +struct Fixture { + temp: Temp, + _server: Server, +} +impl Fixture { + fn new(work: Value, tolerance: u32, spread: u32) -> Self { + let temp = Temp::new(); + save(temp.path("work.json"), &work); + save(temp.path("deployment.json"), &deployment()); + save( + temp.path("policy.json"), + &declaration(&work, tolerance, spread), + ); + let root = temp.0.clone(); + let requests = work["cells"] + .as_array() + .unwrap() + .iter() + .map(|c| { + c["concurrency"].as_u64().unwrap() + * (c["trials"].as_u64().unwrap() + c["warmup_trials"].as_u64().unwrap()) + }) + .sum::() as usize; + let server = Server::new(move |mut stream, index, _| { + let role = ["a", "b", "a2"][index / requests]; + let policy = fs::read(root.join("policy.json")).unwrap(); + assert_eq!( + fs::read(root.join(role).join("policy.json")).unwrap(), + policy + ); + assert_eq!( + value(root.join(role).join("plan.json"))["policy_sha256"], + hash(&policy) + ); + header(&mut stream, "text/event-stream"); + frame( + &mut stream, + json!({"id":"fixture","choices":[{"delta":{"content":format!("synthetic-{index}")}}]}), + ); + finish(&mut stream, Some(8), Some(0)); + }); + for (index, role) in ["a", "b", "a2"].into_iter().enumerate() { + successful(&collect(&temp, &server, role, true)); + // Explicit synthetic declared starts avoid clock-resolution premises. + rewrite_plan(&temp.path(role), |p| { + p["started_unix_ms"] = json!((index + 1) * 1000) + }); + synthetic_times(&temp.path(role), &[(10000, 20000)]); + } + Self { + temp, + _server: server, + } + } + fn decide(&self, reference: Option<&str>) -> Output { + let mut command = cli(); + command + .arg("decide") + .arg(self.temp.path("a")) + .arg(self.temp.path("b")) + .arg("--json"); + if let Some(reference) = reference { + command.arg("--reference").arg(self.temp.path(reference)); + } + command.output().unwrap() + } +} +// Synthetic evidence transformations retain the real CLI's request/body verifier. +// They are numeric fixtures, not an assertion that the edited times were observed. +fn rewrite_plan(root: &Path, change: impl FnOnce(&mut Value)) { + let mut plan = value(root.join("plan.json")); + change(&mut plan); + save(root.join("plan.json"), &plan); + let plan_hash = hash(&fs::read(root.join("plan.json")).unwrap()); + for spec in plan["waves"].as_array().unwrap() { + let dir = root.join(format!("wave-{:06}", spec["index"].as_u64().unwrap())); + if !dir.exists() { + continue; + } + let mut reservation = value(dir.join("reservation.json")); + reservation["plan_sha256"] = json!(plan_hash); + save(dir.join("reservation.json"), &reservation); + if dir.join("wave.json").exists() { + let mut wave = value(dir.join("wave.json")); + wave["plan_sha256"] = json!(plan_hash); + wave["reservation_sha256"] = + json!(hash(&fs::read(dir.join("reservation.json")).unwrap())); + save(dir.join("wave.json"), &wave); + } + } + let session_path = root.join("session-000000/session.json"); + let mut session = value(&session_path); + session["plan_sha256"] = json!(plan_hash); + session["collector_sha256"] = plan["collector_sha256"].clone(); + save(session_path, &session); +} +fn synthetic_times(root: &Path, times: &[(u64, u64)]) { + let plan = value(root.join("plan.json")); + for (index, spec) in plan["waves"].as_array().unwrap().iter().enumerate() { + let path = root.join(format!( + "wave-{:06}/wave.json", + spec["index"].as_u64().unwrap() + )); + let mut wave = value(&path); + let (first, settle) = times[index % times.len()]; + for attempt in wave["attempts"].as_array_mut().unwrap() { + attempt["timing"] = json!({"dispatch_offset_us":0,"headers_us":0,"first_body_us":0, + "first_generated_text_us":first,"first_generated_channel":"answer", + "first_answer_text_us":first,"settle_us":settle,"capture_parse_us":0}); + } + wave["elapsed_us"] = json!(settle); + wave["dispatch_spread_us"] = json!(0); + wave["achieved_completion_tokens_per_second"] = + json!(wave["completion_tokens"].as_u64().unwrap() as f64 * 1_000_000.0 / settle as f64); + save(path, &wave); + } +} +fn report(output: &Output, outcome: &str, exit: i32) -> Value { + assert_eq!( + output.status.code(), + Some(exit), + "{}", + String::from_utf8_lossy(&output.stderr) + ); + let value: Value = serde_json::from_slice(&output.stdout).unwrap(); + assert_eq!(value["version"], 1); + assert_eq!(value["decision"], outcome); + value +} +fn gate<'a>(report: &'a Value, metric: &str) -> &'a Value { + report["gates"] + .as_array() + .unwrap() + .iter() + .find(|g| g["metric"] == metric) + .unwrap() +} +fn reason(value: &Value, expected: &str) { + assert!( + value["reason_codes"] + .as_array() + .unwrap() + .contains(&json!(expected)), + "{value}" + ); +} + +#[test] +fn policy_capture_all_roles_and_exact_zero_tolerance_determinism() { + let fixture = Fixture::new(workload(2, 1, 3), 0, 0); + let first = fixture.decide(Some("a2")); + let decision = report(&first, "PASS", 0); + assert_eq!(first.stdout, fixture.decide(Some("a2")).stdout); + for metric in METRICS { + assert_eq!(gate(&decision, metric)["decision"], "PASS"); + } + assert_eq!( + gate(&decision, METRICS[2])["coverage"]["candidate"]["expected_observations"], + 6 + ); + synthetic_times(&fixture.temp.path("b"), &[(10001, 20002)]); + let changed = report(&fixture.decide(Some("a2")), "REGRESSION", 3); + for metric in METRICS { + assert_eq!(gate(&changed, metric)["decision"], "REGRESSION"); + } + assert_ne!( + decision["roles"]["candidate"]["evidence_sha256"], + changed["roles"]["candidate"]["evidence_sha256"] + ); +} + +#[test] +fn policy_exact_9999_latency_and_rate_boundaries() { + let fixture = Fixture::new(workload(1, 1, 3), 9999, 0); + for role in ["a", "a2"] { + synthetic_times(&fixture.temp.path(role), &[(50, 100)]); + } + synthetic_times(&fixture.temp.path("b"), &[(50, 199)]); + let before = report(&fixture.decide(Some("a2")), "PASS", 0); + assert_eq!(gate(&before, METRICS[0])["decision"], "PASS"); + // Integer latency equality: 19999 / 10000 = 1 + 9999 / 10000. + for role in ["a", "a2"] { + synthetic_times(&fixture.temp.path(role), &[(5000, 10000)]); + } + synthetic_times(&fixture.temp.path("b"), &[(5000, 19999)]); + report(&fixture.decide(Some("a2")), "PASS", 0); + synthetic_times(&fixture.temp.path("b"), &[(5000, 20000)]); + let over = report(&fixture.decide(Some("a2")), "REGRESSION", 3); + assert_eq!(gate(&over, METRICS[0])["decision"], "REGRESSION"); + for role in ["a", "a2"] { + synthetic_times(&fixture.temp.path(role), &[(50, 100)]); + } + synthetic_times(&fixture.temp.path("b"), &[(500000, 1000000)]); + let exact = report(&fixture.decide(Some("a2")), "REGRESSION", 3); + for metric in &METRICS[1..] { + assert_eq!(gate(&exact, metric)["decision"], "PASS"); + } + synthetic_times(&fixture.temp.path("b"), &[(500001, 1000002)]); + let over = report(&fixture.decide(Some("a2")), "REGRESSION", 3); + for metric in &METRICS[1..] { + assert_eq!(gate(&over, metric)["decision"], "REGRESSION"); + } +} + +#[test] +fn policy_reference_spread_and_straddling_are_not_pass() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 10000); + synthetic_times(&fixture.temp.path("a2"), &[(20000, 40000)]); + synthetic_times(&fixture.temp.path("b"), &[(15000, 30000)]); + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + reason(gate(&decision, METRICS[1]), "envelope_straddles_tolerance"); + synthetic_times(&fixture.temp.path("a2"), &[(20001, 40002)]); + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + for metric in METRICS { + reason(gate(&decision, metric), "reference_spread_exceeded"); + } +} + +#[test] +fn policy_missing_reused_out_of_order_and_unqualified_reference() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + reason( + &report(&fixture.decide(None), "INCONCLUSIVE", 2), + "missing_reference", + ); + reason( + &report(&fixture.decide(Some("a")), "INCONCLUSIVE", 2), + "role_reuse", + ); + rewrite_plan(&fixture.temp.path("a2"), |p| { + p["started_unix_ms"] = json!(1500) + }); + reason( + &report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2), + "declared_starts_out_of_order", + ); + rewrite_plan(&fixture.temp.path("a2"), |p| { + p["started_unix_ms"] = json!(3000); + p["deployment"]["settings"] = json!("private-different-setting"); + }); + let output = fixture.decide(Some("a2")); + reason(&report(&output, "INCONCLUSIVE", 2), "reference_unqualified"); + assert!( + !String::from_utf8(output.stdout) + .unwrap() + .contains("private-different-setting") + ); +} + +#[test] +fn policy_partial_lane_and_warmup_coverage_preserves_other_regression() { + let fixture = Fixture::new(workload(2, 1, 3), 0, 0); + synthetic_times(&fixture.temp.path("b"), &[(10001, 20002)]); + let path = fixture.temp.path("b/wave-000001/wave.json"); + let mut wave = value(&path); + // Prefill is undefined at a zero first-text interval, but latency is measured. + wave["attempts"][0]["timing"]["first_generated_text_us"] = json!(0); + wave["attempts"][0]["timing"]["first_answer_text_us"] = json!(0); + save(&path, &wave); + let decision = report(&fixture.decide(Some("a2")), "REGRESSION", 3); + reason(gate(&decision, METRICS[3]), "metric_unavailable"); + assert_eq!( + gate(&decision, METRICS[3])["coverage"]["candidate"]["observed_observations"], + 5 + ); + assert_eq!(gate(&decision, METRICS[0])["decision"], "REGRESSION"); + let no_warmup = Fixture::new(workload(1, 0, 3), 0, 0); + let decision = report(&no_warmup.decide(Some("a2")), "INCONCLUSIVE", 2); + for metric in METRICS { + reason(gate(&decision, metric), "warmup_incomplete"); + } +} + +#[test] +fn policy_all_cells_and_metrics_remain_in_scope() { + let mut work = workload(1, 1, 3); + let mut second = work["cells"][0].clone(); + second["id"] = json!("second"); + work["cells"].as_array_mut().unwrap().push(second); + let fixture = Fixture::new(work, 0, 0); + let decision = report(&fixture.decide(Some("a2")), "PASS", 0); + let scope = decision["gates"] + .as_array() + .unwrap() + .iter() + .map(|g| (g["cell"].as_str().unwrap(), g["metric"].as_str().unwrap())) + .collect::>(); + for cell in ["cell", "second"] { + for metric in METRICS { + assert!(scope.contains(&(cell, metric))); + } + } +} + +#[test] +fn policy_sidecar_raw_body_and_outcome_fingerprints() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + let before = report(&fixture.decide(Some("a2")), "PASS", 0); + let path = fixture.temp.path("b/wave-000001/response-0000.bin"); + let bytes = fs::read(&path).unwrap(); + let changed = String::from_utf8(bytes) + .unwrap() + .replace("synthetic-", "synthetic-edited-"); + fs::write(&path, &changed).unwrap(); + report(&fixture.decide(Some("a2")), "ERROR", 1); + let receipt = fixture.temp.path("b/wave-000001/wave.json"); + let mut wave = value(&receipt); + let delta = changed.len() - wave["attempts"][0]["response_bytes"].as_u64().unwrap() as usize; + wave["attempts"][0]["response_bytes"] = json!(changed.len()); + wave["attempts"][0]["response_sha256"] = json!(hash(changed.as_bytes())); + wave["attempts"][0]["terminal_offset"] = + json!(wave["attempts"][0]["terminal_offset"].as_u64().unwrap() + delta as u64); + save(receipt, &wave); + let after = report(&fixture.decide(Some("a2")), "PASS", 0); + assert_ne!( + before["roles"]["candidate"]["evidence_sha256"], + after["roles"]["candidate"]["evidence_sha256"] + ); + let mut outcome = value(fixture.temp.path("b/run.json")); + outcome["wave_preparation_us"] = json!(outcome["wave_preparation_us"].as_u64().unwrap() + 1); + save(fixture.temp.path("b/run.json"), &outcome); + save(fixture.temp.path("b/session-000000/run.json"), &outcome); + let end = report(&fixture.decide(Some("a2")), "PASS", 0); + assert_ne!( + after["roles"]["candidate"]["evidence_sha256"], + end["roles"]["candidate"]["evidence_sha256"] + ); + let policy_path = fixture.temp.path("a2/policy.json"); + let mut bytes = fs::read(&policy_path).unwrap(); + bytes.push(b' '); + fs::write(&policy_path, &bytes).unwrap(); + reason( + &report(&fixture.decide(Some("a2")), "ERROR", 1), + "policy_hash_mismatch", + ); + rewrite_plan(&fixture.temp.path("a2"), |p| { + p["policy_sha256"] = json!(hash(&bytes)) + }); + reason( + &report(&fixture.decide(Some("a2")), "ERROR", 1), + "conflicting_policy", + ); + fs::remove_file(&policy_path).unwrap(); + reason( + &report(&fixture.decide(Some("a2")), "ERROR", 1), + "invalid_policy", + ); +} + +#[test] +fn policy_absent_candidate_cannot_be_bound_retrospectively_and_errors_are_private() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + rewrite_plan(&fixture.temp.path("b"), |p| { + p.as_object_mut().unwrap().remove("policy_sha256"); + }); + reason( + &report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2), + "missing_policy", + ); + fs::write( + fixture.temp.path("b/plan.json"), + b"private-loader-diagnostic-prompt", + ) + .unwrap(); + let output = fixture.decide(Some("a2")); + reason(&report(&output, "ERROR", 1), "invalid_evidence"); + assert!( + !String::from_utf8(output.stdout) + .unwrap() + .contains("private-loader-diagnostic-prompt") + ); +} + +#[test] +fn policy_admission_rejects_schema_scope_pins_and_trial_minima_before_dispatch() { + let temp = Temp::new(); + let server = Server::new(normal); + let work = workload(1, 1, 3); + save(temp.path("work.json"), &work); + save(temp.path("deployment.json"), &deployment()); + let policy = declaration(&work, 0, 0); + let mut invalid = Vec::new(); + for (field, value) in [ + ("version", json!(2)), + ("method", json!("unknown")), + ("unknown", json!(true)), + ("collector_sha256", json!("0".repeat(64))), + ("workload_source_sha256", json!("0".repeat(64))), + ("min_trials", json!(4)), + ("cells", json!([])), + ] { + let mut p = policy.clone(); + p[field] = value; + invalid.push(p); + } + let mut duplicate = policy.clone(); + duplicate["cells"][0]["metrics"][1] = policy["cells"][0]["metrics"][0].clone(); + invalid.push(duplicate); + let mut duplicate_cell = policy.clone(); + duplicate_cell["cells"] + .as_array_mut() + .unwrap() + .push(policy["cells"][0].clone()); + invalid.push(duplicate_cell); + for (field, value) in [ + ("cell", json!("unknown-cell")), + ("metrics", json!([])), + ("unknown", json!(true)), + ] { + let mut p = policy.clone(); + p["cells"][0][field] = value; + invalid.push(p); + } + for (field, value) in [ + ("metric", json!("unknown")), + ("max_regression_bps", json!(10000)), + ("max_reference_spread_bps", json!(1000001)), + ("unknown", json!(0)), + ] { + let mut p = policy.clone(); + p["cells"][0]["metrics"][0][field] = value; + invalid.push(p); + } + for (index, policy) in invalid.iter().enumerate() { + save(temp.path("policy.json"), policy); + let name = format!("rejected-{index}"); + assert_eq!(collect(&temp, &server, &name, true).status.code(), Some(1)); + assert!(!temp.path(&name).exists()); + } + fs::write(temp.path("policy.json"), vec![b' '; 65537]).unwrap(); + assert_eq!( + collect(&temp, &server, "oversized", true).status.code(), + Some(1) + ); + assert_eq!(server.count.load(Ordering::SeqCst), 0); +} + +#[test] +fn policy_legacy_default_roundtrip_is_unbound_not_pass() { + let temp = Temp::new(); + let server = Server::new(normal); + successful(&run(&temp, &server, "old", &workload(1, 0, 1))); + let paths = [ + "plan.json", + "workload.json", + "run.json", + "wave-000000/reservation.json", + "wave-000000/wave.json", + "wave-000000/response-0000.bin", + ]; + let before: Vec<_> = paths + .iter() + .map(|p| fs::read(temp.path("old").join(p)).unwrap()) + .collect(); + assert!( + value(temp.path("old/plan.json")) + .get("policy_sha256") + .is_none() + ); + let compare = cli() + .arg("compare") + .arg(temp.path("old")) + .arg(temp.path("old")) + .arg("--json") + .output() + .unwrap(); + successful(&compare); + let decision = cli() + .arg("decide") + .arg(temp.path("old")) + .arg(temp.path("old")) + .arg("--json") + .output() + .unwrap(); + reason(&report(&decision, "INCONCLUSIVE", 2), "missing_policy"); + for (path, bytes) in paths.iter().zip(before) { + assert_eq!(fs::read(temp.path("old").join(path)).unwrap(), bytes); + } +} + +fn synthetic_usage(root: &Path, tokens: impl Fn(usize) -> u64) { + let plan = value(root.join("plan.json")); + for spec in plan["waves"] + .as_array() + .unwrap() + .iter() + .filter(|s| s["phase"] == "measured") + { + let dir = root.join(format!("wave-{:06}", spec["index"].as_u64().unwrap())); + let mut wave = value(dir.join("wave.json")); + let mut total = 0; + for (lane, attempt) in wave["attempts"] + .as_array_mut() + .unwrap() + .iter_mut() + .enumerate() + { + let tokens = tokens(lane); + total += tokens; + let usage = json!({"prompt_tokens":4,"completion_tokens":tokens,"total_tokens":tokens+4, + "prompt_tokens_details":{"cached_tokens":0}}); + let path = dir.join(format!("response-{lane:04}.bin")); + let original = fs::read_to_string(&path).unwrap(); + let raw = original + .split_inclusive('\n') + .map(|line| { + if let Some(data) = line.strip_prefix("data: ") + && let Ok(mut event) = serde_json::from_str::(data) + && event.get("usage").is_some() + { + event["usage"] = usage.clone(); + return format!("data: {event}\n"); + } + line.to_owned() + }) + .collect::(); + fs::write(&path, &raw).unwrap(); + let delta = raw.len() as i64 - original.len() as i64; + attempt["response_bytes"] = json!(raw.len()); + attempt["response_sha256"] = json!(hash(raw.as_bytes())); + attempt["terminal_offset"] = + json!(attempt["terminal_offset"].as_i64().unwrap() + delta); + attempt["usage"]["completion_tokens"] = json!(tokens); + attempt["usage"]["total_tokens"] = json!(tokens + 4); + } + wave["completion_tokens"] = json!(total); + wave["achieved_completion_tokens_per_second"] = + json!(total as f64 * 1_000_000.0 / wave["elapsed_us"].as_u64().unwrap() as f64); + save(dir.join("wave.json"), &wave); + } +} + +#[test] +fn policy_defined_zero_throughput_is_not_an_undefined_or_matched_rate() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + synthetic_usage(&fixture.temp.path("b"), |_| 0); + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + let aggregate = gate(&decision, METRICS[1]); + reason(aggregate, "output_amounts_mismatch"); + assert_eq!( + aggregate["coverage"]["candidate"]["observed_observations"], + 3 + ); + assert_eq!(aggregate["ranges"]["candidate"][0]["numerator"], 0); + assert_eq!( + gate(&decision, METRICS[2])["coverage"]["candidate"]["observed_observations"], + 0 + ); + for role in ["a", "a2"] { + synthetic_usage(&fixture.temp.path(role), |_| 0); + } + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + reason(gate(&decision, METRICS[1]), "nonpositive_reference"); + assert_eq!(gate(&decision, METRICS[0])["decision"], "PASS"); +} + +#[test] +fn policy_ordered_lane_amounts_and_decode_dropout_are_not_flattened() { + let fixture = Fixture::new(workload(2, 1, 3), 0, 0); + for role in ["a", "b", "a2"] { + synthetic_usage( + &fixture.temp.path(role), + |lane| if lane == 0 { 1 } else { 8 }, + ); + } + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + reason(gate(&decision, METRICS[2]), "metric_unavailable"); + assert_eq!( + gate(&decision, METRICS[2])["coverage"]["candidate"]["observed_observations"], + 3 + ); + assert_eq!(gate(&decision, METRICS[1])["decision"], "PASS"); + synthetic_usage( + &fixture.temp.path("a2"), + |lane| if lane == 0 { 8 } else { 1 }, + ); + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + for metric in METRICS { + reason(gate(&decision, metric), "output_amounts_mismatch"); + } +} + +#[test] +fn policy_loader_uses_recorded_collector_not_evaluator_binary() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + let before = report(&fixture.decide(Some("a2")), "PASS", 0); + let mut policy = value(fixture.temp.path("policy.json")); + policy["collector_sha256"] = json!("1".repeat(64)); + let bytes = serde_json::to_vec(&policy).unwrap(); + for role in ["a", "b", "a2"] { + fs::write(fixture.temp.path(role).join("policy.json"), &bytes).unwrap(); + rewrite_plan(&fixture.temp.path(role), |p| { + p["collector_sha256"] = policy["collector_sha256"].clone(); + p["policy_sha256"] = json!(hash(&bytes)); + }); + } + let decision = report(&fixture.decide(Some("a2")), "PASS", 0); + assert_eq!(decision["evaluator_sha256"], before["evaluator_sha256"]); + assert_ne!( + decision["evaluator_sha256"], + decision["roles"]["baseline"]["collector_sha256"] + ); + rewrite_plan(&fixture.temp.path("b"), |p| { + p["collector_sha256"] = json!("2".repeat(64)) + }); + reason( + &report(&fixture.decide(Some("a2")), "ERROR", 1), + "policy_collector_mismatch", + ); + policy["workload_source_sha256"] = json!("3".repeat(64)); + let bytes = serde_json::to_vec(&policy).unwrap(); + fs::write(fixture.temp.path("b/policy.json"), &bytes).unwrap(); + rewrite_plan(&fixture.temp.path("b"), |p| { + p["collector_sha256"] = policy["collector_sha256"].clone(); + p["policy_sha256"] = json!(hash(&bytes)); + }); + reason( + &report(&fixture.decide(Some("a2")), "ERROR", 1), + "policy_source_mismatch", + ); +} + +#[test] +fn policy_copied_lineage_cannot_be_relabelled_as_an_acquisition() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + fn copy(source: &Path, destination: &Path) { + fs::create_dir(destination).unwrap(); + for entry in fs::read_dir(source).unwrap() { + let entry = entry.unwrap(); + if entry.file_type().unwrap().is_dir() { + copy(&entry.path(), &destination.join(entry.file_name())); + } else { + fs::copy(entry.path(), destination.join(entry.file_name())).unwrap(); + } + } + } + copy(&fixture.temp.path("a"), &fixture.temp.path("copy")); + rewrite_plan(&fixture.temp.path("copy"), |p| { + p["started_unix_ms"] = json!(4000) + }); + reason( + &report(&fixture.decide(Some("copy")), "INCONCLUSIVE", 2), + "role_reuse", + ); +} + +#[test] +fn policy_incomplete_session_and_missing_wave_cannot_pass() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + fs::remove_file(fixture.temp.path("a2/run.json")).unwrap(); + reason( + &report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2), + "session_incomplete", + ); + fs::remove_file(fixture.temp.path("a2/session-000000/run.json")).unwrap(); + fs::remove_dir_all(fixture.temp.path("a2/wave-000003")).unwrap(); + let decision = report(&fixture.decide(Some("a2")), "INCONCLUSIVE", 2); + reason(gate(&decision, METRICS[0]), "metric_unavailable"); + assert_eq!( + gate(&decision, METRICS[0])["coverage"]["reference"]["observed_waves"], + 2 + ); +} + +#[test] +fn policy_resume_reuses_capture_after_external_declaration_changes() { + let temp = Temp::new(); + let work = workload(1, 1, 3); + save(temp.path("work.json"), &work); + save(temp.path("deployment.json"), &deployment()); + save(temp.path("policy.json"), &declaration(&work, 0, 0)); + let captured = fs::read(temp.path("policy.json")).unwrap(); + let (ready, releases) = std::sync::mpsc::sync_channel(1); + let root = temp.path("run"); + let expected = captured.clone(); + let server = Server::new(move |mut stream, index, _| { + assert_eq!(fs::read(root.join("policy.json")).unwrap(), expected); + if index == 0 { + let (release, wait) = std::sync::mpsc::sync_channel(0); + ready.send(release).unwrap(); + wait.recv_timeout(Duration::from_secs(30)).unwrap(); + } + header(&mut stream, "text/event-stream"); + frame( + &mut stream, + json!({"id":"fixture","choices":[{"delta":{"content":"synthetic-resume"}}]}), + ); + finish(&mut stream, Some(8), Some(0)); + }); + let mut child = cli() + .arg("run") + .arg(temp.path("work.json")) + .args([ + "--endpoint", + &server.endpoint, + "--model", + "fixture-model", + "--local-http", + "--json", + ]) + .arg("--policy") + .arg(temp.path("policy.json")) + .arg("--out") + .arg(temp.path("run")) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + .unwrap(); + server.wait_for_request(&mut child); + successful(&cli().arg("pause").arg(temp.path("run")).output().unwrap()); + releases + .recv_timeout(Duration::from_secs(30)) + .unwrap() + .send(()) + .unwrap(); + let paused = child.wait_with_output().unwrap(); + assert_eq!(paused.status.code(), Some(2)); + assert_eq!( + serde_json::from_slice::(&paused.stdout).unwrap()["status"], + "paused" + ); + let plan = fs::read(temp.path("run/plan.json")).unwrap(); + fs::write(temp.path("policy.json"), b"invalid replacement declaration").unwrap(); + let resumed = cli() + .arg("resume") + .arg(temp.path("run")) + .arg("--json") + .output() + .unwrap(); + assert_eq!(resumed.status.code(), Some(2)); + assert_eq!( + serde_json::from_slice::(&resumed.stdout).unwrap()["status"], + "completed-with-ineligible-measurements" + ); + assert_eq!(server.count.load(Ordering::SeqCst), 4); + assert_eq!(fs::read(temp.path("run/policy.json")).unwrap(), captured); + assert_eq!(fs::read(temp.path("run/plan.json")).unwrap(), plan); + let output = cli() + .arg("decide") + .arg(temp.path("run")) + .arg(temp.path("run")) + .arg("--reference") + .arg(temp.path("run")) + .arg("--json") + .output() + .unwrap(); + reason(&report(&output, "INCONCLUSIVE", 2), "session_incomplete"); +} + +#[test] +fn policy_unbound_collector_identity_is_validated_before_emission() { + let fixture = Fixture::new(workload(1, 1, 3), 0, 0); + let marker = format!("{:x<64}", "private://synthetic-sensitive"); + for role in ["a", "b", "a2"] { + rewrite_plan(&fixture.temp.path(role), |plan| { + plan.as_object_mut().unwrap().remove("policy_sha256"); + plan["collector_sha256"] = json!(marker); + }); + fs::remove_file(fixture.temp.path(role).join("policy.json")).unwrap(); + } + let output = fixture.decide(Some("a2")); + let decision = report(&output, "ERROR", 1); + reason(&decision, "invalid_evidence"); + assert!(!String::from_utf8(output.stdout).unwrap().contains(&marker)); + for role in ["baseline", "candidate", "reference"] { + assert_eq!(decision["roles"].get(role), Some(&Value::Null)); + } +} diff --git a/docs/performance/CONTRACT.md b/docs/performance/CONTRACT.md index c8133db..aa7f02e 100644 --- a/docs/performance/CONTRACT.md +++ b/docs/performance/CONTRACT.md @@ -4,6 +4,29 @@ Workloads and wave receipts remain v1. New execution plans are v2, adding mandat execution-session provenance without changing request rendering or wave timing. Legacy v1 plans remain readable offline but cannot be paused or resumed. +Plans may additionally bind `policy.json` through optional `policy_sha256`. +The field is omitted when unused; existing unbound plans retain their bytes. +Policy-bearing plans require a reader that supports this declaration. + +## Captured policy decisions + +`run --policy FILE` captures and validates the same exact pre-dispatch policy +for baseline, candidate and repeat. It changes neither generation requests nor +ordinary measurement eligibility. Offline `decide` verifies the saved evidence +and policy bindings; it never applies a new policy to unbound historical runs. + +The decision envelope is version 1, separate from comparison JSON version 3. +PASS means complete observed evidence satisfies every required policy gate; +REGRESSION means a qualified observed adverse bound exceeds the declared +tolerance. INCONCLUSIVE preserves missing or insufficient evidence and excessive +reference variability; ERROR denotes invalid or unverifiable input. These are +descriptive policy outcomes, not significance, causality or universal guarantees. +An eligible comparison or successful process exit alone is not a decision. + +The [policy guide](README.md#captured-observed-envelope-policy) defines the closed +schema, exact checked rational arithmetic, complete-coverage requirements, +reference-spread gate, outcome precedence and command-specific exits. + ## Workload admission The workload has `version`, `name`, `request`, `limits`, `cases` and `cells`. diff --git a/docs/performance/README.md b/docs/performance/README.md index a29805d..77ae842 100644 --- a/docs/performance/README.md +++ b/docs/performance/README.md @@ -4,6 +4,119 @@ It does not start servers, download models, grade answers or require quality packs. Its `run`, `pause`, `resume` and offline `compare` workflows are separate from `grill`. +## Captured observed-envelope policy + +`run WORKLOAD --policy FILE ...` validates a bounded policy declaration before +dispatch and saves its exact bytes as `policy.json` before publishing the plan. +Every baseline, candidate and baseline repeat must capture the same policy. +Omitting the option preserves the legacy plan, request and receipt encoding; +an unbound run cannot acquire a policy retrospectively through `decide`. +`resume` reads the captured declaration, with no replacement-policy option. +Policy-bearing plans require this reader; older readers may reject the new +optional `policy_sha256` field. + +The declaration is a closed JSON object, bounded to 64 KiB: + +```json +{ + "version": 1, + "method": "observed-envelope-v1", + "id": "approved-workload-envelope", + "collector_sha256": "REPLACE_WITH_ACTUAL_BINARY_SHA256", + "workload_source_sha256": "REPLACE_WITH_EXACT_WORKLOAD_FILE_SHA256", + "min_trials": 3, + "cells": [ + { + "cell": "REPLACE_WITH_WORKLOAD_CELL_ID", + "metrics": [ + { + "metric": "wave_latency_us", + "max_regression_bps": 0, + "max_reference_spread_bps": 0 + } + ] + } + ] +} +``` + +Replace the illustrative pins and cell with real values before admission; both +pins must be lowercase SHA-256 hex. Enumerate every workload cell exactly once. +Each cell declares a nonempty unique subset of `wave_latency_us`, +`achieved_completion_tokens_per_second`, `decode_tokens_per_second` and +`prefill_tokens_per_second`. Metric direction is intrinsic. `min_trials` is +3–100 and cannot exceed any cell's declared measured trials. Decisions require +at least one declared warmup per cell. Regression tolerances are integer basis +points in 0–9999; the observed reference-spread budget is in 0–1,000,000. +100 basis points equals one percent. These are schema bounds, not recommended +scientific margins; the zero values above illustrate strict equality budgets, +not a generally suitable operating policy. + +Collection checks the pins against the exact workload source and actual +collector executable. Offline loading checks the captured pins against each +recorded plan, not the evaluator's current executable. The decision separately +identifies that evaluator. Policy bytes, not normalized JSON, determine binding. +Present corrupt or conflicting bindings produce `ERROR`; missing bindings or a +missing repeat produce `INCONCLUSIVE`, never a retrospective application. + +```sh +target/release/grill-perf decide results/baseline results/candidate \ + --reference results/baseline-repeat --json +``` + +This offline command requires distinct, ordered baseline/candidate/repeat +acquisitions, currently qualified `declared_match` baseline/repeat declarations, +complete eligible uninterrupted sessions, complete warmups and measured trials, +and equal ordered trial/lane completion amounts. Copied evidence under another +path does not establish a distinct acquisition. Starts and deployment +declarations do not authenticate nonoverlap, restoration or independent +execution. Operators remain responsible for approval before collection and for +restoring the baseline without overlapping runs. + +Every declared gate remains in the result, including unavailable metrics and +partial lanes. Raw verified observations form rational pairs: latency is +`(elapsed_us, 1)`, aggregate throughput is `(completion_tokens, elapsed_us)`, +decode is `(completion_tokens - 1, settle_us - first_generated_text_us)` and +prefill is `(prompt_tokens, first_generated_text_us)`. Availability and cache +rules match the descriptive measurements. Serialized rate ranges use these +tokens-per-microsecond pairs; multiplying by 1,000,000 gives tokens/second. +Zero or undefined intervals and nonpositive reference minima cannot qualify. +A defined zero candidate throughput is not silently dropped, but still must +satisfy the output-amount and eligibility requirements. + +Baseline and repeat extrema are pooled as `[L,U]`, candidate extrema as `[l,u]`. +The reference variability gate requires `U/L - 1` within the declared spread +budget. Latency adverse bounds are `[l/U - 1, u/L - 1]`; rate-loss bounds are +`[1 - u/L, 1 - l/U]`. Negative adverse bounds describe observed improvement. +The decision uses checked integer cross-products, not rounded percentages or +floating-point epsilon: `PASS` requires the worst bound within tolerance; +`REGRESSION` requires the best bound strictly beyond tolerance; otherwise the +gate is `INCONCLUSIVE`. Floating adverse bounds are descriptive only. +Arithmetic overflow produces `ERROR`. Reference spread is an operator-selected +observed variability gate, not a confidence bound. + +Aggregate precedence is `ERROR`, `REGRESSION`, `INCONCLUSIVE`, then `PASS`. +A qualified regression is not erased by another gate's uncertainty. `PASS` +requires every gate, not a favorable filtered subset. No outcome establishes +statistical significance, causal effect, universal no-regression or live +qualification. + +The versioned decision JSON separates eligibility from policy outcome and +includes policy identity, evaluator identity, role-labelled verified evidence +hashes, scope, expected/observed coverage, rational ranges, tolerances and bounded +reason codes. Fingerprints bind exact plan/workload, session states and ordered +wave/raw-response/companion evidence, including missing states. They are not +execution attestations. Identical verified inputs and evaluator yield identical +JSON without a new timestamp, filesystem paths or raw private diagnostics. +Raw evidence and existing descriptive comparison output remain private by +default; a decision summary is not replay evidence. + +Only a successfully printed versioned decision envelope constitutes a decision. +For `decide`, exits are `PASS` 0, `ERROR` 1, `INCONCLUSIVE` 2 and `REGRESSION` 3. +Parsing or output failures can share exit values without producing a decision. +Existing `compare` exit 0 remains an eligibility-only result, never `PASS`; +other commands retain their exit semantics. + ## Build and use Requirements match the workspace: Linux, Rust/Cargo 1.98 and the native build