metrale_bench/benchmarks/scheduler_equivalence/
compare.rs

1// SPDX-License-Identifier: MIT OR Apache-2.0
2
3//! 2026-09-26: Given what each pass produced, decide whether the synchronous
4//! and asynchronous routers are the same function of a request. No I/O, so
5//! the tests build each failure without a server.
6//!
7//! Owner: bench, scheduler-equivalence gate.
8//! Invariants: `verdict` passes only when every cell has samples, not every
9//! reference reply is empty, nothing is unmeasured, no control diverged and
10//! no async reply diverged.
11
12use std::collections::BTreeMap;
13use std::time::Duration;
14
15use super::host::{Lane, Router};
16use super::unmeasured::{self, Unmeasured};
17use crate::benchmarks::transcript::Transcript;
18use crate::http::RequestFailure;
19
20/// 2026-09-26: Which of the three passes over a lane a leg is.
21#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
22pub enum Pass {
23    /// 2026-09-26: The reference: the synchronous serve.
24    Sync,
25    /// 2026-09-26: The candidate: the asynchronous serve.
26    Async,
27    /// 2026-09-26: The synchronous serve measured a second time: what "the
28    /// same router answers the same twice" looks like on this workload.
29    Control,
30}
31
32impl Pass {
33    pub fn label(self) -> &'static str {
34        match self {
35            Pass::Sync => "sync",
36            Pass::Async => "async",
37            Pass::Control => "sync-control",
38        }
39    }
40    pub fn router(self) -> Router {
41        match self {
42            Pass::Sync | Pass::Control => Router::Sync,
43            Pass::Async => Router::Async,
44        }
45    }
46}
47
48/// 2026-09-26: What one sample produced in one leg.
49#[derive(Debug, Clone)]
50pub struct Reply {
51    pub sample_id: String,
52    /// 2026-09-26: A failure is its own variant, so two failed requests can
53    /// never compare equal and read as agreement.
54    pub outcome: Result<Box<Transcript>, RequestFailure>,
55    /// 2026-09-26: Issue to last byte or failure, on the gate's clock.
56    pub elapsed: Duration,
57}
58
59/// 2026-09-26: One full pass over the draw: a lane, a pass, a concurrency.
60#[derive(Debug, Clone)]
61pub struct Leg {
62    pub lane: Lane,
63    pub pass: Pass,
64    pub concurrency: usize,
65    pub replies: Vec<Reply>,
66}
67
68impl Leg {
69    pub(super) fn find(&self, sample_id: &str) -> Option<&Reply> {
70        self.replies.iter().find(|r| r.sample_id == sample_id)
71    }
72}
73
74/// 2026-09-26: One sample, one leg against the reference.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub enum SampleVerdict {
77    /// 2026-09-26: Byte-identical: text, reasoning, tool calls, finish reason
78    /// and the server's completion token count.
79    Equal,
80    /// 2026-09-26: The replies differ. Against the async leg this is the
81    /// finding; against the control it makes an async difference
82    /// unattributable.
83    Diverged {
84        /// 2026-09-26: Bytes of the reference's canonical form that matched
85        /// before the first difference.
86        common_prefix: usize,
87    },
88    /// 2026-09-26: A request failed or a sample is missing: evidence of
89    /// nothing, never counted as agreement.
90    Unmeasured(String),
91}
92
93/// 2026-09-26: The reduction of one leg against the reference leg of its
94/// cell.
95#[derive(Debug, Clone, Default, PartialEq, Eq)]
96pub struct Diff {
97    pub equal: usize,
98    pub diverged: usize,
99    pub unmeasured: usize,
100    /// 2026-09-26: At most `NAMED_DIVERGENCES` ids, for the verdict and the
101    /// report table.
102    pub diverged_ids: Vec<String>,
103}
104
105/// 2026-09-26: One (lane, concurrency) cell: the reference and what was held
106/// against it.
107#[derive(Debug, Clone, PartialEq, Eq)]
108pub struct Cell {
109    pub lane: Lane,
110    pub concurrency: usize,
111    pub samples: usize,
112    /// 2026-09-26: Reference replies whose canonical form has no character at
113    /// or above U+0020. The vacuity guard: an engine answering nothing agrees
114    /// with itself.
115    pub empty_replies: usize,
116    pub async_vs_sync: Diff,
117    /// 2026-09-26: `None` when no control leg was run for the cell.
118    pub control_vs_sync: Option<Diff>,
119    /// 2026-09-26: One entry per request of this cell that failed or never
120    /// came back, in any leg. A failed sync reference is one entry, though it
121    /// leaves the sample unmeasured in both the async and the control diff.
122    pub unmeasured: Vec<Unmeasured>,
123}
124
125#[derive(Debug, Clone, Default)]
126pub struct Score {
127    pub cells: Vec<Cell>,
128}
129
130/// 2026-09-26: How many diverged ids a verdict names before it stops.
131const NAMED_DIVERGENCES: usize = 8;
132
133/// 2026-09-26: The comparison rule, the same as `kat_equality::verdict_for`:
134/// the canonical transcript and the server's token count.
135pub fn verdict_for(sample_id: &str, reference: &Leg, other: &Leg) -> SampleVerdict {
136    let Some(base) = reference.find(sample_id) else {
137        return SampleVerdict::Unmeasured(format!("{sample_id} missing from the reference"));
138    };
139    let base_t = match &base.outcome {
140        Ok(t) => t,
141        Err(e) => return SampleVerdict::Unmeasured(format!("reference: {e}")),
142    };
143    let Some(obs) = other.find(sample_id) else {
144        return SampleVerdict::Unmeasured(format!(
145            "{sample_id} missing from {}",
146            other.pass.label()
147        ));
148    };
149    let t = match &obs.outcome {
150        Ok(t) => t,
151        Err(e) => {
152            return SampleVerdict::Unmeasured(format!("{}: {e}", other.pass.label()));
153        }
154    };
155    let (a, b) = (base_t.canonical(), t.canonical());
156    if a != b || base_t.completion_tokens != t.completion_tokens {
157        return SampleVerdict::Diverged {
158            common_prefix: a.bytes().zip(b.bytes()).take_while(|(x, y)| x == y).count(),
159        };
160    }
161    SampleVerdict::Equal
162}
163
164fn diff(reference: &Leg, other: &Leg) -> Diff {
165    let mut d = Diff::default();
166    for r in &reference.replies {
167        match verdict_for(&r.sample_id, reference, other) {
168            SampleVerdict::Equal => d.equal += 1,
169            SampleVerdict::Diverged { .. } => {
170                d.diverged += 1;
171                if d.diverged_ids.len() < NAMED_DIVERGENCES {
172                    d.diverged_ids.push(r.sample_id.clone());
173                }
174            }
175            SampleVerdict::Unmeasured(_) => d.unmeasured += 1,
176        }
177    }
178    d
179}
180
181fn empty(t: &Transcript) -> bool {
182    t.canonical().chars().all(|c| (c as u32) < 0x20)
183}
184
185/// 2026-09-26: Group the legs into cells and reduce each against its sync
186/// reference. A cell with no sync reference has zero samples, which the
187/// verdict refuses; a cell with no async leg counts every sample unmeasured.
188pub fn score(legs: &[Leg]) -> Score {
189    let mut keys: Vec<(Lane, usize)> = legs.iter().map(|l| (l.lane, l.concurrency)).collect();
190    keys.sort();
191    keys.dedup();
192    let find = |lane: Lane, c: usize, pass: Pass| {
193        legs.iter()
194            .find(|l| l.lane == lane && l.concurrency == c && l.pass == pass)
195    };
196    let mut cells = Vec::new();
197    for (lane, c) in keys {
198        let reference = find(lane, c, Pass::Sync);
199        let candidate = find(lane, c, Pass::Async);
200        let control = find(lane, c, Pass::Control);
201        let unmeasured = reference
202            .map(|r| unmeasured::collect(r, candidate, control))
203            .unwrap_or_default();
204        let (samples, empty_replies, async_vs_sync, control_vs_sync) = match reference {
205            Some(r) => (
206                r.replies.len(),
207                r.replies
208                    .iter()
209                    .filter(|x| matches!(&x.outcome, Ok(t) if empty(t)))
210                    .count(),
211                // 2026-09-26: A missing candidate leg is every sample
212                // unmeasured, never an empty diff that reads as agreement.
213                candidate.map(|a| diff(r, a)).unwrap_or(Diff {
214                    unmeasured: r.replies.len(),
215                    ..Diff::default()
216                }),
217                control.map(|k| diff(r, k)),
218            ),
219            None => (0, 0, Diff::default(), None),
220        };
221        cells.push(Cell {
222            lane,
223            concurrency: c,
224            samples,
225            empty_replies,
226            async_vs_sync,
227            control_vs_sync,
228            unmeasured,
229        });
230    }
231    Score { cells }
232}
233
234/// 2026-09-26: The gate's rule, in order. Anything not proven equal fails,
235/// and a control that diverged makes the router's difference unattributable.
236pub fn verdict(s: &Score) -> crate::result::Verdict {
237    use crate::result::Verdict;
238    if s.cells.is_empty() {
239        return Verdict::fail(
240            "EQUIVALENCE UNPROVEN: no (lane, concurrency) cell ran, so nothing was compared."
241                .to_string(),
242        );
243    }
244    for c in &s.cells {
245        let where_ = format!("{} C={}", c.lane.label(), c.concurrency);
246        if c.samples == 0 {
247            return Verdict::fail(format!(
248                "EQUIVALENCE UNPROVEN: the sync reference at {where_} issued no samples."
249            ));
250        }
251        if c.empty_replies == c.samples {
252            return Verdict::fail(format!(
253                "EQUIVALENCE VACUOUS: all {} replies at {where_} were empty, so the routers \
254                 agree trivially. This proves the harness ran, not that the routers are \
255                 equivalent.",
256                c.samples
257            ));
258        }
259        if c.async_vs_sync.unmeasured > 0 {
260            return Verdict::fail(format!(
261                "EQUIVALENCE UNPROVEN: {} of {} samples at {where_} were unmeasured in the \
262                 async pass (a failed request is not evidence of agreement). Causes: {}.",
263                c.async_vs_sync.unmeasured,
264                c.samples,
265                unmeasured::breakdown(&c.unmeasured)
266            ));
267        }
268        if let Some(k) = &c.control_vs_sync {
269            if k.unmeasured > 0 {
270                return Verdict::fail(format!(
271                    "EQUIVALENCE UNPROVEN: {} of {} samples at {where_} were unmeasured in \
272                     the sync control pass. Causes: {}.",
273                    k.unmeasured,
274                    c.samples,
275                    unmeasured::breakdown(&c.unmeasured)
276                ));
277            }
278            if k.diverged > 0 {
279                return Verdict::fail(format!(
280                    "CONTROL DIVERGED: the synchronous router answered {} of {} samples \
281                     differently on a second pass at {where_} — the workload is not \
282                     deterministic under this lane, so an async difference there would be \
283                     unattributable ({}{}). Pin the lane harder or find the nondeterminism \
284                     before reading the async column.",
285                    k.diverged,
286                    c.samples,
287                    k.diverged_ids.join(", "),
288                    more(k)
289                ));
290            }
291        }
292    }
293    let failing: Vec<&Cell> = s
294        .cells
295        .iter()
296        .filter(|c| c.async_vs_sync.diverged > 0)
297        .collect();
298    if !failing.is_empty() {
299        let per_cell: Vec<String> = failing
300            .iter()
301            .map(|c| {
302                format!(
303                    "{} C={}: {} of {} ({}{})",
304                    c.lane.label(),
305                    c.concurrency,
306                    c.async_vs_sync.diverged,
307                    c.samples,
308                    c.async_vs_sync.diverged_ids.join(", "),
309                    more(&c.async_vs_sync)
310                )
311            })
312            .collect();
313        return Verdict::fail(format!(
314            "ROUTER-DEPENDENT: samples changed between the synchronous and the asynchronous \
315             router while the control held — {}. At temperature 0 a reply must be a function \
316             of its request, whichever router served it.",
317            per_cell.join("; ")
318        ));
319    }
320    let lanes: Vec<&str> = {
321        let mut v: Vec<&str> = s.cells.iter().map(|c| c.lane.label()).collect();
322        v.dedup();
323        v
324    };
325    let cs: Vec<String> = {
326        let mut v: Vec<usize> = s.cells.iter().map(|c| c.concurrency).collect();
327        v.sort_unstable();
328        v.dedup();
329        v.iter().map(|c| c.to_string()).collect()
330    };
331    let controlled = s.cells.iter().all(|c| c.control_vs_sync.is_some());
332    Verdict::pass(format!(
333        "{} samples byte-identical between sync and async at C={{{}}} under {} ({})",
334        s.cells.first().map(|c| c.samples).unwrap_or(0),
335        cs.join(","),
336        lanes.join(" and "),
337        if controlled {
338            "sync-vs-sync control held in every cell"
339        } else {
340            "no control leg"
341        }
342    ))
343}
344
345fn more(d: &Diff) -> String {
346    let left = d.diverged.saturating_sub(d.diverged_ids.len());
347    if left > 0 {
348        format!(" +{left} more")
349    } else {
350        String::new()
351    }
352}
353
354/// 2026-09-26: Raw numbers for the record. Every class is a key even at zero.
355pub fn metrics(s: &Score, diagnostics: &BTreeMap<String, f64>) -> BTreeMap<String, f64> {
356    let mut m = BTreeMap::new();
357    let sum = |f: &dyn Fn(&Cell) -> usize| s.cells.iter().map(f).sum::<usize>() as f64;
358    m.insert("cells".into(), s.cells.len() as f64);
359    m.insert(
360        "samples".into(),
361        s.cells.first().map(|c| c.samples).unwrap_or(0) as f64,
362    );
363    m.insert("diverged".into(), sum(&|c| c.async_vs_sync.diverged));
364    m.insert(
365        "unmeasured".into(),
366        sum(&|c| {
367            c.async_vs_sync.unmeasured
368                + c.control_vs_sync
369                    .as_ref()
370                    .map(|k| k.unmeasured)
371                    .unwrap_or(0)
372        }),
373    );
374    m.insert(
375        "control_diverged".into(),
376        sum(&|c| c.control_vs_sync.as_ref().map(|k| k.diverged).unwrap_or(0)),
377    );
378    m.insert("empty_replies".into(), sum(&|c| c.empty_replies));
379    m.extend(unmeasured::metrics(
380        s.cells.iter().flat_map(|c| c.unmeasured.iter()),
381    ));
382    for c in &s.cells {
383        m.insert(
384            format!("diverged_{}_c{}", c.lane.label(), c.concurrency),
385            c.async_vs_sync.diverged as f64,
386        );
387        m.insert(
388            format!("control_diverged_{}_c{}", c.lane.label(), c.concurrency),
389            c.control_vs_sync.as_ref().map(|k| k.diverged).unwrap_or(0) as f64,
390        );
391    }
392    for (k, v) in diagnostics {
393        m.insert(format!("async_{k}"), *v);
394    }
395    m
396}
metrale_bench/benchmarks/scheduler_equivalence/compare.rs

metrale_bench/benchmarks/scheduler_equivalence/
compare.rs
