metrale_bench/benchmarks/contamination/
driver.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: The contamination state machine. Each `next()` runs one step
4//! (endpoint probe, a leg, or scoring), so the pane updates as each leg lands
5//! and a cancellation is seen before the next step.
6//!
7//! Legs, in order (the design is in `score.rs`):
8//! 1. prime: every probe once, solo, unmeasured, so later legs can run
9//!    cache-warm. A failed prime is logged as a warning.
10//! 2. two references: every probe solo, twice. A probe whose two runs
11//!    disagree is `AloneUnstable`.
12//! 3. rungs: per configured concurrency, one batch with the probes in its
13//!    first slots and copies of the probes as ballast (see `rung_slots`).
14//! 4. post: every probe solo again. A divergence here is `Persistent`.
15//!
16//! A transport failure becomes [`RequestOutcome::Error`], which the scorer
17//! classifies as `Unmeasured`; any unmeasured cell fails the verdict.
18//!
19//! Owner: bench (contamination).
20//! Invariants: none beyond the types.
21
22use crate::hardware::Sensitivity;
23use std::future::Future;
24use std::time::{Duration, Instant};
25
26use anyhow::{Context, Result, bail};
27use serde_json::{Value, json};
28
29use crate::benchmark::{Benchmark, BenchmarkDescriptor};
30use crate::benchmarks::one_line;
31use crate::http;
32use crate::metadata::PluginMetadata;
33use crate::params::{ParamKind, ParamSpec, ParamValue, ParamValues};
34use crate::plugin::{Plugin, PluginHandle};
35use crate::result::{BenchmarkResult, LogLine, RunStatus, Verdict};
36
37use super::prompts::{self, Probe};
38use super::report;
39use super::score::{Legs, Score, score, verdict};
40use super::transcript::{RequestOutcome, Transcript};
41
42const SUMMARY: &str = "Concurrent requests must not change each other's output";
43pub const METADATA: PluginMetadata = PluginMetadata::metrale(SUMMARY);
44
45pub const DESCRIPTOR: BenchmarkDescriptor = BenchmarkDescriptor {
46    id: "cross-contamination",
47    name: "Cross-Contamination Detector",
48    summary: SUMMARY,
49    detail: "Four legs at temperature 0: prime every probe cache-warm, record each probe's solo \
50             output twice, re-run the probes inside concurrent batches at several concurrency \
51             rungs, then re-run them solo afterwards. Each probe carries a lexical canary; a \
52             foreign canary in a reply is leakage on its own evidence, any other token-level \
53             difference from the solo reference is a divergence, and a divergence in the closing \
54             solo leg means corrupted state SURVIVED the batch. Zero tolerance — token identity \
55             has no noise term to allow for.",
56    duration_hint: "~2–5 min",
57    expected_secs: 240,
58    updated: "2026-08-09",
59    needs_confirmation: false,
60    // 2026-09-26: Any served model is a valid subject: this tests the engine,
61    // not a checkpoint.
62    intended_for: None,
63    threshold_params: &[],
64    // 2026-09-26: Token identity does not depend on clock speed, so a
65    // throttled box gives the same result.
66    sensitivity: Sensitivity::Correctness,
67    ctor: || Box::new(CrossContamination::default()),
68};
69
70#[derive(Default, Clone, Copy, PartialEq, Eq, Debug)]
71enum Phase {
72    #[default]
73    Prime,
74    RefA,
75    RefB,
76    Rung,
77    Post,
78    Score,
79    Done,
80}
81
82#[derive(Default)]
83pub struct CrossContamination {
84    handle: Option<PluginHandle>,
85    phase: Phase,
86    concurrencies: Vec<usize>,
87    rung_cursor: usize,
88    ref_a: Vec<RequestOutcome>,
89    ref_b: Vec<RequestOutcome>,
90    rungs: Vec<(String, Vec<RequestOutcome>)>,
91    post: Vec<RequestOutcome>,
92    max_tokens: usize,
93    min_completion_tokens: usize,
94    timeout: Duration,
95    started: Option<Instant>,
96    probed: bool,
97}
98
99/// 2026-09-26: The request every leg sends: streamed, at temperature 0,
100/// because the scorer compares transcripts for equality.
101pub(super) fn request_body(model: &str, prompt: &str, max_tokens: usize) -> Value {
102    json!({
103        "model": model,
104        "stream": true,
105        "temperature": 0.0,
106        "max_tokens": max_tokens,
107        "messages": [{"role": "user", "content": prompt}],
108    })
109}
110
111/// 2026-09-26: Which probe each slot of a rung batch runs. The first
112/// `n_probes` slots are the measured ones, in probe order, because
113/// [`super::score::Legs`] indexes outcomes by probe. The rest cycle through
114/// the probes as unmeasured ballast. A batch never has fewer than `n_probes`
115/// slots.
116pub(super) fn rung_slots(n_probes: usize, conc: usize) -> Vec<usize> {
117    (0..conc.max(n_probes)).map(|s| s % n_probes).collect()
118}
119
120/// 2026-09-26: An error becomes [`RequestOutcome::Error`] rather than an
121/// abort, so one failed request does not discard the finished legs.
122pub(super) fn to_outcome(r: Result<http::ChatOutcome>) -> RequestOutcome {
123    match r {
124        Ok(o) => RequestOutcome::Ok(Box::new(Transcript::from(&o))),
125        Err(e) => RequestOutcome::Error(one_line(format!("{e:#}"))),
126    }
127}
128
129impl CrossContamination {
130    fn handle(&self) -> Result<&PluginHandle> {
131        self.handle.as_ref().context("benchmark was not loaded")
132    }
133
134    fn elapsed(&self) -> Duration {
135        self.started.map(|s| s.elapsed()).unwrap_or_default()
136    }
137
138    /// 2026-09-26: prime, two references, one per rung, post, score.
139    fn total_steps(&self) -> u64 {
140        self.concurrencies.len() as u64 + 5
141    }
142
143    fn steps_done(&self) -> u64 {
144        match self.phase {
145            Phase::Prime => 0,
146            Phase::RefA => 1,
147            Phase::RefB => 2,
148            Phase::Rung => 3 + self.rung_cursor as u64,
149            Phase::Post => 3 + self.concurrencies.len() as u64,
150            Phase::Score => 4 + self.concurrencies.len() as u64,
151            Phase::Done => self.total_steps(),
152        }
153    }
154
155    async fn one(&self, probe: &Probe) -> RequestOutcome {
156        let Ok(handle) = self.handle() else {
157            return RequestOutcome::Error("benchmark was not loaded".into());
158        };
159        let target = handle.target();
160        let body = request_body(&target.model, &probe.prompt(), self.max_tokens);
161        to_outcome(http::chat_stream(target, &body, self.timeout).await)
162    }
163
164    /// 2026-09-26: Every probe once, one request at a time.
165    async fn solo_leg(&self, label: &str) -> Result<Vec<RequestOutcome>> {
166        let handle = self.handle()?.clone();
167        let mut out = Vec::with_capacity(prompts::PROBES.len());
168        for probe in &prompts::PROBES {
169            handle.check_cancelled()?;
170            handle.status(format!("{label} · probe {} solo", probe.name));
171            out.push(self.one(probe).await);
172        }
173        Ok(out)
174    }
175
176    /// 2026-09-26: The whole batch in flight at once; only the first
177    /// `PROBES.len()` outcomes are kept (see [`rung_slots`]).
178    async fn run_rung(&self, conc: usize) -> Vec<RequestOutcome> {
179        let slots = rung_slots(prompts::PROBES.len(), conc);
180        let futures: Vec<_> = slots
181            .iter()
182            .map(|&p| self.one(&prompts::PROBES[p]))
183            .collect();
184        let mut outcomes = futures::future::join_all(futures).await;
185        outcomes.truncate(prompts::PROBES.len());
186        outcomes
187    }
188
189    /// 2026-09-26: Scores the recorded legs. Reads only collected state, so
190    /// the tests call it without a server.
191    pub(super) fn scored(&self) -> (Score, Verdict) {
192        let canaries = prompts::canaries();
193        let legs = Legs {
194            ref_a: &self.ref_a,
195            ref_b: &self.ref_b,
196            rungs: &self.rungs,
197            post: &self.post,
198            canaries: &canaries,
199            min_completion_tokens: self.min_completion_tokens,
200        };
201        let s = score(&legs);
202        let v = verdict(&s);
203        (s, v)
204    }
205
206    fn frame(&self, phase: &str, line: Option<LogLine>) -> BenchmarkResult {
207        let mut f = BenchmarkResult::running(phase, self.elapsed())
208            .with_progress(self.steps_done(), self.total_steps());
209        if let Some(line) = line {
210            f = f.log_line(line);
211        }
212        f
213    }
214}
215
216impl Plugin for CrossContamination {
217    fn metadata(&self) -> &'static PluginMetadata {
218        &METADATA
219    }
220
221    fn load(&mut self, handle: PluginHandle) -> impl Future<Output = Result<()>> + Send {
222        self.handle = Some(handle);
223        self.started = Some(Instant::now());
224        async { Ok(()) }
225    }
226}
227
228impl Benchmark for CrossContamination {
229    fn descriptor(&self) -> &'static BenchmarkDescriptor {
230        &DESCRIPTOR
231    }
232
233    fn parameters(&self) -> Vec<ParamSpec> {
234        vec![
235            ParamSpec::new(
236                "concurrencies",
237                "Concurrency rungs",
238                "Batch sizes for the concurrent leg; the probes ride inside each batch.",
239                ParamKind::IntList { min: 2, max: 128 },
240                ParamValue::IntList(vec![2, 4, 8]),
241            ),
242            ParamSpec::new(
243                "max_tokens",
244                "Max tokens",
245                "Output budget per request. The probes answer in well under this.",
246                ParamKind::Int { min: 32, max: 4096 },
247                ParamValue::Int(256),
248            ),
249            ParamSpec::new(
250                "min_completion_tokens",
251                "Completion-token floor",
252                "Replies shorter than this are Unmeasured: two empty replies are equal and prove nothing.",
253                ParamKind::Int { min: 1, max: 4096 },
254                ParamValue::Int(16),
255            ),
256            ParamSpec::new(
257                "request_timeout_s",
258                "Request timeout",
259                "Seconds before a single request is abandoned and scored as Unmeasured.",
260                ParamKind::Int { min: 10, max: 3600 },
261                ParamValue::Int(300),
262            ),
263        ]
264    }
265
266    fn configure(&mut self, values: &ParamValues) -> Result<()> {
267        let specs = self.parameters();
268        values.validate_against(&specs)?;
269        let floor = values.usize("min_completion_tokens")?;
270        let max_tokens = values.usize("max_tokens")?;
271        // 2026-09-26: A floor at or above the budget would make every reply
272        // Unmeasured, so it is rejected here, before any request.
273        if floor >= max_tokens {
274            bail!(
275                "Completion-token floor: {floor} must be below the max_tokens budget \
276                 ({max_tokens}), or every reply is Unmeasured by construction"
277            );
278        }
279        self.concurrencies = values
280            .int_list("concurrencies")?
281            .iter()
282            .map(|c| *c as usize)
283            .collect();
284        self.max_tokens = max_tokens;
285        self.min_completion_tokens = floor;
286        self.timeout = Duration::from_secs(values.usize("request_timeout_s")? as u64);
287        self.phase = Phase::Prime;
288        self.probed = false;
289        self.rung_cursor = 0;
290        self.ref_a.clear();
291        self.ref_b.clear();
292        self.rungs.clear();
293        self.post.clear();
294        Ok(())
295    }
296
297    async fn next(&mut self) -> Result<BenchmarkResult> {
298        let handle = self.handle()?.clone();
299        handle.check_cancelled()?;
300
301        if !self.probed {
302            self.probed = true;
303            http::probe(handle.target(), Duration::from_secs(10))
304                .await
305                .context("endpoint probe failed — check the target URL and port")?;
306            if self.concurrencies.is_empty() {
307                bail!("no concurrency rungs configured");
308            }
309            return Ok(self.frame(
310                "probe",
311                Some(LogLine::info(format!(
312                    "{} · model {} · {} probes × {} rungs",
313                    handle.target().base_url,
314                    handle.target().model,
315                    prompts::PROBES.len(),
316                    self.concurrencies.len()
317                ))),
318            ));
319        }
320
321        match self.phase {
322            Phase::Prime => {
323                let primes = self.solo_leg("prime").await?;
324                let failed = primes
325                    .iter()
326                    .filter(|o| matches!(o, RequestOutcome::Error(_)))
327                    .count();
328                self.phase = Phase::RefA;
329                let line = if failed > 0 {
330                    LogLine::warn(format!(
331                        "{failed} prime request(s) failed — later legs may run cache-cold"
332                    ))
333                } else {
334                    LogLine::info("primed — all measured legs run cache-warm")
335                };
336                Ok(self.frame("prime", Some(line)))
337            }
338            Phase::RefA => {
339                self.ref_a = self.solo_leg("reference 1/2").await?;
340                self.phase = Phase::RefB;
341                Ok(self.frame("reference 1/2", None))
342            }
343            Phase::RefB => {
344                self.ref_b = self.solo_leg("reference 2/2").await?;
345                self.phase = Phase::Rung;
346                Ok(self.frame("reference 2/2", None))
347            }
348            Phase::Rung => {
349                let conc = self.concurrencies[self.rung_cursor];
350                handle.status(format!("concurrency {conc} · batch in flight"));
351                let measured = self.run_rung(conc).await;
352                self.rungs.push((format!("c{conc}"), measured));
353                self.rung_cursor += 1;
354                if self.rung_cursor >= self.concurrencies.len() {
355                    self.phase = Phase::Post;
356                }
357                handle.progress(self.steps_done(), self.total_steps());
358                Ok(self.frame(format!("concurrency {conc}").as_str(), None))
359            }
360            Phase::Post => {
361                self.post = self.solo_leg("post-check").await?;
362                self.phase = Phase::Score;
363                Ok(self.frame("post-check", None))
364            }
365            Phase::Score => {
366                let (s, v) = self.scored();
367                self.phase = Phase::Done;
368                let line = LogLine::info(one_line(format!(
369                    "{} comparisons: {} identical · {} diverged · {} contaminated · \
370                     {} persistent · {} unmeasured",
371                    s.compared, s.identical, s.diverged, s.contaminated, s.persistent, s.unmeasured
372                )));
373                Ok(BenchmarkResult {
374                    status: RunStatus::Completed,
375                    ..BenchmarkResult::running("done", self.elapsed())
376                }
377                .with_progress(self.total_steps(), self.total_steps())
378                .with_summary(report::summary(&s))
379                .with_table(report::table(&s))
380                .with_metrics(report::metrics(&s))
381                .with_verdict(v)
382                .log_line(line))
383            }
384            Phase::Done => bail!("next() was called after the run finished"),
385        }
386    }
387}
388
389#[cfg(test)]
390#[path = "driver_tests.rs"]
391mod driver_tests;
metrale_bench/benchmarks/contamination/driver.rs

metrale_bench/benchmarks/contamination/
driver.rs
