loadmaster_probe/
run.rs

1//! Execute fixtures against an endpoint and reduce them to a verdict.
2
3use std::path::PathBuf;
4
5use serde::Serialize;
6use serde_json::{Map, Value, json};
7
8use crate::check::{eval, query};
9use crate::client::{Client, Exchange};
10use crate::detect::Detected;
11use crate::fixture::{Fixture, Kind, parse_ref};
12use crate::known::{KnownIssue, matches};
13
14pub const PROBE_VERSION: &str = env!("CARGO_PKG_VERSION");
15
16#[derive(Debug, Clone, Default)]
17pub struct Options {
18    pub model: String,
19    /// Override every fixture's repeat count.
20    pub repeat: Option<u32>,
21    /// blake3 of the chat template the endpoint should use (from the model file).
22    pub expect_template: Option<String>,
23    /// Write each request and raw response here.
24    pub capture_dir: Option<PathBuf>,
25    /// Context for failure signatures, when known (e.g. from models.lock).
26    pub quant: Option<String>,
27}
28
29#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
30#[serde(tag = "result", content = "detail", rename_all = "snake_case")]
31pub enum Outcome {
32    Pass,
33    Fail(String),
34    Skip(String),
35}
36
37#[derive(Debug, Clone, Serialize)]
38pub struct FixtureResult {
39    pub id: String,
40    pub description: String,
41    pub core: bool,
42    pub outcome: Outcome,
43    /// Each run's outcome; the fixture passes only if all pass.
44    pub runs: Vec<Outcome>,
45    pub latency_ms: u64,
46}
47
48#[derive(Debug, Clone, Serialize)]
49pub struct Report {
50    pub probe_version: String,
51    pub url: String,
52    pub model: String,
53    pub detected: Detected,
54    pub quant: Option<String>,
55    pub template_blake3: Option<String>,
56    pub results: Vec<FixtureResult>,
57    /// `pass`, `partial` or `fail`.
58    pub verdict: String,
59    /// Failure classes, sorted and deduplicated.
60    pub failures: Vec<String>,
61    pub known_issues: Vec<KnownIssue>,
62}
63
64impl Report {
65    /// 0 pass, 1 partial, 2 fail.
66    pub fn exit_code(&self) -> u8 {
67        match self.verdict.as_str() {
68            "pass" => 0,
69            "partial" => 1,
70            _ => 2,
71        }
72    }
73}
74
75/// Replace `{"$from": "N:<path>"}` objects with values from earlier exchanges.
76fn substitute(v: &Value, xs: &[Exchange]) -> Result<Value, String> {
77    match v {
78        Value::Object(o) if o.len() == 1 && o.contains_key("$from") => {
79            let r = parse_ref(o["$from"].as_str().ok_or("`$from` takes a string")?)?;
80            query(xs, &r)
81                .into_iter()
82                .next()
83                .filter(|v| !v.is_null())
84                .ok_or_else(|| format!("{} not found in earlier responses", o["$from"]))
85        }
86        Value::Object(o) => {
87            let mut m = Map::new();
88            for (k, x) in o {
89                m.insert(k.clone(), substitute(x, xs)?);
90            }
91            Ok(Value::Object(m))
92        }
93        Value::Array(a) => a
94            .iter()
95            .map(|x| substitute(x, xs))
96            .collect::<Result<_, _>>()
97            .map(Value::Array),
98        other => Ok(other.clone()),
99    }
100}
101
102fn request_body(step_body: &Value, stream: bool, model: &str) -> Value {
103    let mut body = json!({ "model": model, "temperature": 0, "seed": 42, "max_tokens": 512 });
104    if let Some(o) = step_body.as_object() {
105        for (k, v) in o {
106            body[k] = v.clone();
107        }
108    }
109    body["stream"] = json!(stream);
110    body
111}
112
113fn capture(dir: &Option<PathBuf>, fixture: &str, run: u32, step: usize, x: &Exchange) {
114    let Some(dir) = dir else { return };
115    let d = dir.join(fixture);
116    if std::fs::create_dir_all(&d).is_err() {
117        return;
118    }
119    let stem = format!("run{run}-step{step}");
120    let _ = std::fs::write(
121        d.join(format!("{stem}.request.json")),
122        serde_json::to_vec_pretty(&x.request).unwrap_or_default(),
123    );
124    let _ = std::fs::write(d.join(format!("{stem}.response.raw")), &x.raw);
125}
126
127async fn run_once(c: &Client, f: &Fixture, opts: &Options, run: u32) -> (Outcome, u64) {
128    let mut xs: Vec<Exchange> = Vec::new();
129    let mut latency = 0;
130    for (i, step) in f.steps.iter().enumerate() {
131        let Ok(body) = substitute(&step.body, &xs) else {
132            break; // an earlier step did not produce what this one needs
133        };
134        let x = c.chat(request_body(&body, step.stream, &opts.model)).await;
135        latency += x.latency_ms;
136        capture(&opts.capture_dir, &f.id, run, i, &x);
137        xs.push(x);
138    }
139    if xs.is_empty() {
140        return (Outcome::Fail("no_response".into()), latency);
141    }
142    if xs.iter().any(|x| x.status == 0) {
143        return (Outcome::Fail("unreachable".into()), latency);
144    }
145    if eval(&f.pass, &xs) {
146        return (Outcome::Pass, latency);
147    }
148    let class = f
149        .fail
150        .iter()
151        .find(|r| eval(&r.when, &xs))
152        .map_or_else(|| "unclassified".to_owned(), |r| r.class.clone());
153    (Outcome::Fail(class), latency)
154}
155
156fn template_check(detected: &Detected, opts: &Options) -> Outcome {
157    let Some(want) = &opts.expect_template else {
158        return Outcome::Skip(
159            "no expected template; pass --expect-template or run via `loadmaster up --verify`"
160                .into(),
161        );
162    };
163    match &detected.chat_template {
164        None => Outcome::Skip(format!(
165            "{} does not report its chat template",
166            detected.runtime
167        )),
168        Some(t) if blake3::hash(t.as_bytes()).to_hex().as_str() == want => Outcome::Pass,
169        Some(_) => Outcome::Fail("template_mismatch".into()),
170    }
171}
172
173pub async fn run_fixture(
174    c: &Client,
175    f: &Fixture,
176    opts: &Options,
177    detected: &Detected,
178) -> FixtureResult {
179    let (outcome, runs, latency_ms) = match f.kind {
180        Kind::Template => {
181            let o = template_check(detected, opts);
182            (o.clone(), vec![o], 0)
183        }
184        Kind::Chat => {
185            let n = opts.repeat.unwrap_or(f.repeat).max(1);
186            let mut runs = Vec::new();
187            let mut latency = 0;
188            for run in 0..n {
189                let (o, l) = run_once(c, f, opts, run).await;
190                latency += l;
191                runs.push(o);
192            }
193            let first = runs[0].clone();
194            let outcome = if runs.iter().all(|r| *r == first) {
195                first
196            } else {
197                // Disagreeing runs at temperature 0: report the first failure seen.
198                runs.iter().find(|r| matches!(r, Outcome::Fail(_))).map_or(
199                    Outcome::Fail("nondeterministic".into()),
200                    |r| match r {
201                        Outcome::Fail(c) => Outcome::Fail(format!(
202                            "{c} (in {} of {n} runs)",
203                            runs.iter().filter(|x| *x == r).count()
204                        )),
205                        other => other.clone(),
206                    },
207                )
208            };
209            (outcome, runs, latency / u64::from(n))
210        }
211    };
212    FixtureResult {
213        id: f.id.clone(),
214        description: f.description.clone(),
215        core: f.core,
216        outcome,
217        runs,
218        latency_ms,
219    }
220}
221
222/// Verdict rules (spec §6): `pass` when nothing failed, `partial` when every core
223/// fixture passed, `fail` otherwise. Skipped fixtures do not count either way.
224pub fn verdict(results: &[FixtureResult]) -> &'static str {
225    let failed = |r: &&FixtureResult| matches!(r.outcome, Outcome::Fail(_));
226    if !results.iter().any(|r| failed(&r)) {
227        "pass"
228    } else if results
229        .iter()
230        .filter(|r| r.core)
231        .all(|r| r.outcome == Outcome::Pass)
232        && results.iter().any(|r| r.core)
233    {
234        "partial"
235    } else {
236        "fail"
237    }
238}
239
240pub async fn probe(
241    c: &Client,
242    fixtures: &[Fixture],
243    opts: &Options,
244    known: &[KnownIssue],
245) -> Report {
246    let detected = crate::detect::detect(c).await;
247    let mut results = Vec::new();
248    for f in fixtures {
249        results.push(run_fixture(c, f, opts, &detected).await);
250    }
251    let mut failures: Vec<String> = results
252        .iter()
253        .filter_map(|r| match &r.outcome {
254            Outcome::Fail(c) => Some(c.split(' ').next().unwrap_or(c).to_owned()),
255            _ => None,
256        })
257        .collect();
258    failures.sort();
259    failures.dedup();
260    let known_issues = known
261        .iter()
262        .filter(|k| {
263            failures
264                .iter()
265                .any(|f| matches(k, f, &detected, &opts.model))
266        })
267        .cloned()
268        .collect();
269    Report {
270        probe_version: PROBE_VERSION.into(),
271        url: c.base().to_owned(),
272        model: opts.model.clone(),
273        template_blake3: detected
274            .chat_template
275            .as_ref()
276            .map(|t| blake3::hash(t.as_bytes()).to_hex().to_string()),
277        quant: opts.quant.clone(),
278        verdict: verdict(&results).into(),
279        detected,
280        results,
281        failures,
282        known_issues,
283    }
284}
loadmaster_probe/run.rs

loadmaster_probe/
run.rs
