use serde::Serialize;
use serde_json::Value;
use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Instant;
use warble_eval_compare::{compare, CompareRequest, CompareResult, MatchMode, Table, Tolerance};
mod ablation;
mod cache;
mod capture;
mod compliance;
mod context;
mod filter;
mod gate;
pub use ablation::{
format_ablation, run_ablation, AblationConfig, AblationPoint, AblationReport,
StepRecommendation,
};
pub use cache::{rescore, CaseKey, Trace, TraceStore};
pub use capture::{build_candidate_yaml, candidates_header, CaptureInput};
pub use compliance::{
format_compliance, score_compliance, Check, CheckStatus, ComplianceIr, ComplianceReport,
ComplianceTrace, TraceEvent,
};
pub use context::{
classify, compute_mdl_sha, parse_context_version, stamp_context_version, verify_context,
Freshness, VerifyResult,
};
pub use filter::{select_cases, CaseFilter, Sample, Selection};
pub use gate::{format_gate, run_gate, FlakyCase, GateResult, Regression};
#[derive(Debug, serde::Deserialize)]
pub struct Golden {
#[serde(default)]
pub dataset: Option<String>,
#[serde(default)]
pub context_version: Option<String>,
pub cases: Vec<GoldenCase>,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ResultKind {
#[default]
Table,
Verdict,
}
#[derive(Debug, Clone, serde::Deserialize)]
pub struct GoldenCase {
pub id: String,
pub question: String,
#[serde(default)]
pub tags: Vec<String>,
#[serde(rename = "match")]
pub match_mode: MatchMode,
#[serde(default)]
pub tolerance: Tolerance,
pub expected: Table,
#[serde(default)]
pub result_kind: ResultKind,
#[serde(default)]
pub verdict_field: Option<String>,
}
struct SampleOutcome {
pass: bool,
reason: String,
cost: f64,
latency_ms: u64,
turns: u64,
cache_hit: bool,
answer: Option<String>,
}
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct SampleVerdict {
pub pass: bool,
pub reason: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub answer: Option<String>,
}
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct CaseResult {
pub id: String,
pub tags: Vec<String>,
#[serde(default)]
pub samples: u32,
#[serde(default)]
pub passes: u32,
#[serde(default)]
pub pass_rate: f64,
pub pass: bool,
#[serde(default)]
pub flaky: bool,
pub reason: String,
pub cost: f64,
pub latency_ms: u64,
#[serde(default)]
pub turns: u64,
#[serde(default)]
pub cache_hits: u32,
#[serde(default)]
pub cache_misses: u32,
#[serde(default)]
pub samples_detail: Vec<SampleVerdict>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub answer_dist: Option<BTreeMap<String, u32>>,
}
fn fold_samples(
id: String,
tags: Vec<String>,
outcomes: Vec<SampleOutcome>,
record_answers: bool,
) -> CaseResult {
let samples = outcomes.len() as u32;
let passes = outcomes.iter().filter(|o| o.pass).count() as u32;
let pass_rate = if samples > 0 {
passes as f64 / samples as f64
} else {
0.0
};
let pass = samples > 0 && passes == samples;
let flaky = passes > 0 && passes < samples;
let reason = if passes == samples {
"match".to_string()
} else {
modal_reason(
outcomes
.iter()
.filter(|o| !o.pass)
.map(|o| o.reason.as_str()),
)
};
let cost = outcomes.iter().map(|o| o.cost).sum();
let latency_ms = if samples > 0 {
outcomes.iter().map(|o| o.latency_ms).sum::<u64>() / samples as u64
} else {
0
};
let turns = if samples > 0 {
outcomes.iter().map(|o| o.turns).sum::<u64>() / samples as u64
} else {
0
};
let cache_hits = outcomes.iter().filter(|o| o.cache_hit).count() as u32;
let cache_misses = samples - cache_hits;
let answer_dist = record_answers.then(|| {
let mut dist: BTreeMap<String, u32> = BTreeMap::new();
for o in &outcomes {
if let Some(answer) = &o.answer {
*dist.entry(answer.clone()).or_insert(0) += 1;
}
}
dist
});
let samples_detail = outcomes
.into_iter()
.map(|o| SampleVerdict {
pass: o.pass,
reason: o.reason,
answer: if record_answers { o.answer } else { None },
})
.collect();
CaseResult {
id,
tags,
samples,
passes,
pass_rate,
pass,
flaky,
reason,
cost,
latency_ms,
turns,
cache_hits,
cache_misses,
samples_detail,
answer_dist,
}
}
fn modal_reason<'a>(reasons: impl Iterator<Item = &'a str>) -> String {
let mut counts: Vec<(&str, u32)> = Vec::new();
for r in reasons {
match counts.iter_mut().find(|(seen, _)| *seen == r) {
Some(entry) => entry.1 += 1,
None => counts.push((r, 1)),
}
}
let mut best: Option<(&str, u32)> = None;
for (r, c) in counts {
if best.is_none_or(|(_, bc)| c > bc) {
best = Some((r, c));
}
}
best.map(|(r, _)| r.to_string()).unwrap_or_default()
}
fn render_answer(result: &serde_json::Value) -> String {
result.to_string()
}
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct TagStat {
pub pass: u32,
pub n: u32,
#[serde(default)]
pub pass_rate_sum: f64,
}
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct ConfigReport {
pub model: String,
pub n: usize,
pub accuracy: f64,
pub cost_total_usd: f64,
pub latency_ms_avg: u64,
#[serde(default)]
pub turns_avg: u64,
#[serde(default)]
pub cache_hits: usize,
#[serde(default)]
pub cache_misses: usize,
#[serde(default)]
pub flaky_cases: u32,
pub by_tag: BTreeMap<String, TagStat>,
pub cases: Vec<CaseResult>,
}
#[derive(Debug, Clone, Serialize, serde::Deserialize)]
pub struct Report {
pub dataset: Option<String>,
pub context_version: Option<String>,
#[serde(default = "default_parallel")]
pub parallel: usize,
#[serde(default)]
pub selected_cases: usize,
#[serde(default)]
pub total_cases: usize,
pub configs: Vec<ConfigReport>,
}
fn default_parallel() -> usize {
1
}
pub struct RunConfig {
pub project: PathBuf,
pub agent_dir: PathBuf,
pub golden_path: PathBuf,
pub models: Vec<String>,
pub out: Option<PathBuf>,
pub parallel: usize,
pub filter: CaseFilter,
pub no_cache: bool,
pub cache_dir: Option<PathBuf>,
pub samples: usize,
pub record_answers: bool,
}
pub(crate) struct CaseCtx<'a> {
pub model: &'a str,
pub agent_sha: &'a str,
pub context_sha: &'a str,
pub context_version: Option<&'a str>,
pub store: &'a TraceStore,
pub record_answers: bool,
}
impl CaseCtx<'_> {
fn model_override(&self) -> Option<&str> {
(self.model != cache::FRONTMATTER_MODEL).then_some(self.model)
}
}
pub fn extract_result(text: &str) -> Option<Table> {
serde_json::from_value(extract_result_json(text)?).ok()
}
pub fn extract_result_json(text: &str) -> Option<serde_json::Value> {
let candidate = strip_fence(text);
let start = candidate.find('{')?;
let end = candidate.rfind('}')?;
if end <= start {
return None;
}
let value: serde_json::Value = serde_json::from_str(&candidate[start..=end]).ok()?;
if !value.get("rows").map(|r| r.is_array()).unwrap_or(false) {
return None;
}
Some(value)
}
pub fn extract_verdict_json(text: &str) -> Option<serde_json::Value> {
let candidate = strip_fence(text);
let start = candidate.find('{')?;
let end = candidate.rfind('}')?;
if end <= start {
return None;
}
let value: serde_json::Value = serde_json::from_str(&candidate[start..=end]).ok()?;
let has_verdict = value.get("verdict").is_some();
let has_blocks = value.get("blocks").map(Value::is_array).unwrap_or(false);
if !has_verdict && !has_blocks {
return None;
}
Some(value)
}
pub fn project_verdict_field(envelope: &serde_json::Value, field: &str) -> Option<Table> {
match field {
"fresh" => {
let fresh = envelope.get("verdict")?.get("fresh")?.as_bool()?;
Some(Table {
columns: vec!["fresh".to_string()],
rows: vec![vec![Value::Bool(fresh)]],
})
}
"severity" => {
let status_severity =
envelope
.get("blocks")
.and_then(Value::as_array)
.and_then(|blocks| {
blocks
.iter()
.find(|b| b.get("type").and_then(Value::as_str) == Some("status"))
.and_then(|b| b.get("severity"))
.and_then(Value::as_str)
});
let severity = status_severity.or_else(|| {
envelope
.get("verdict")
.and_then(|v| v.get("severity"))
.and_then(Value::as_str)
})?;
Some(Table {
columns: vec!["severity".to_string()],
rows: vec![vec![Value::String(severity.to_string())]],
})
}
_ => None,
}
}
pub fn score_value(result_value: &Value, case: &GoldenCase) -> Option<CompareResult> {
let actual = match case.result_kind {
ResultKind::Table => serde_json::from_value(result_value.clone()).ok()?,
ResultKind::Verdict => {
let field = case.verdict_field.as_deref().unwrap_or("fresh");
project_verdict_field(result_value, field)?
}
};
Some(compare(&CompareRequest {
match_mode: case.match_mode,
tolerance: case.tolerance,
expected: case.expected.clone(),
actual,
}))
}
fn strip_fence(text: &str) -> &str {
let Some(open) = text.find("```") else {
return text;
};
let after = &text[open + 3..];
let content_start = match after.find('\n') {
Some(i) => open + 3 + i + 1,
None => return text,
};
match text[content_start..].find("```") {
Some(close_rel) => &text[content_start..content_start + close_rel],
None => text,
}
}
pub fn aggregate(model: &str, rows: Vec<CaseResult>) -> ConfigReport {
let n = rows.len();
let pass_rate_sum_total: f64 = rows.iter().map(|r| r.pass_rate).sum();
let accuracy = if n > 0 {
pass_rate_sum_total / n as f64
} else {
0.0
};
let flaky_cases = rows.iter().filter(|r| r.flaky).count() as u32;
let cost_total_usd = rows.iter().map(|r| r.cost).sum();
let latency_sum: u64 = rows.iter().map(|r| r.latency_ms).sum();
let latency_ms_avg = if n > 0 { latency_sum / n as u64 } else { 0 };
let turns_sum: u64 = rows.iter().map(|r| r.turns).sum();
let turns_avg = if n > 0 { turns_sum / n as u64 } else { 0 };
let cache_hits = rows.iter().map(|r| r.cache_hits as usize).sum();
let cache_misses = rows.iter().map(|r| r.cache_misses as usize).sum();
let mut by_tag: BTreeMap<String, TagStat> = BTreeMap::new();
for r in &rows {
for t in &r.tags {
let e = by_tag.entry(t.clone()).or_insert(TagStat {
pass: 0,
n: 0,
pass_rate_sum: 0.0,
});
e.n += 1;
if r.pass {
e.pass += 1;
}
e.pass_rate_sum += r.pass_rate;
}
}
ConfigReport {
model: model.to_string(),
n,
accuracy,
cost_total_usd,
latency_ms_avg,
turns_avg,
cache_hits,
cache_misses,
flaky_cases,
by_tag,
cases: rows,
}
}
pub fn format_pareto(report: &Report) -> String {
let mut out = String::new();
out.push_str("\n=== Warble eval — Pareto (accuracy vs cost vs latency) ===\n");
if report.selected_cases < report.total_cases {
out.push_str(&format!(
" (subset: {}/{} golden cases — smoke run, not a full scoring)\n",
report.selected_cases, report.total_cases
));
}
out.push_str(&format!(
"{:<16} {:<7} {:<10} {:<12} {:<7} by_tag\n",
"binding", "acc", "cost($)", "lat(ms)", "turns"
));
for c in &report.configs {
let tags = c
.by_tag
.iter()
.map(|(t, v)| format!("{t}:{:.2}", v.pass_rate_sum / v.n.max(1) as f64))
.collect::<Vec<_>>()
.join(" ");
let cost = if c.cost_total_usd > 0.0 {
format!("{:.4}", c.cost_total_usd)
} else {
"n/a".to_string()
};
out.push_str(&format!(
"{:<16} {:<7} {:<10} {:<12} {:<7} {tags}\n",
format!("strong→{}", c.model),
format!("{:.2}", c.accuracy),
cost,
c.latency_ms_avg,
c.turns_avg
));
}
if report.configs.iter().any(|c| c.cache_hits > 0) {
out.push_str("cache (miss = LLM call this run):\n");
for c in &report.configs {
let note = if c.cache_misses == 0 {
"re-score only, 0 LLM calls this run".to_string()
} else {
format!("{} LLM calls this run", c.cache_misses)
};
out.push_str(&format!(
" strong→{:<10} {} hit / {} miss — {note}\n",
c.model, c.cache_hits, c.cache_misses
));
}
}
if report.configs.iter().any(|c| c.flaky_cases > 0) {
out.push_str("flaky cases (inconsistent across samples):\n");
for c in &report.configs {
for case in c.cases.iter().filter(|case| case.flaky) {
out.push_str(&format!(
" strong→{:<10} {} pass_rate={:.2} ({}/{})",
c.model, case.id, case.pass_rate, case.passes, case.samples
));
if let Some(dist) = &case.answer_dist {
let mut answers: Vec<(&String, &u32)> = dist.iter().collect();
answers.sort_by(|a, b| b.1.cmp(a.1).then_with(|| a.0.cmp(b.0)));
let top = answers
.into_iter()
.take(3)
.map(|(a, n)| format!("{a}×{n}"))
.collect::<Vec<_>>()
.join(", ");
out.push_str(&format!(" — {top}"));
}
out.push('\n');
}
}
}
out
}
struct InstalledAgents {
files: Vec<PathBuf>,
agents_dir: PathBuf,
claude_dir: PathBuf,
}
impl Drop for InstalledAgents {
fn drop(&mut self) {
for f in &self.files {
let _ = fs::remove_file(f);
}
let _ = fs::remove_dir(&self.agents_dir);
let _ = fs::remove_dir(&self.claude_dir);
}
}
pub(crate) fn install_agents(agent_dir: &Path, project: &Path) -> Result<InstalledAgents, String> {
let claude_dir = project.join(".claude");
let agents_dir = claude_dir.join("agents");
fs::create_dir_all(&agents_dir).map_err(|e| format!("mkdir {}: {e}", agents_dir.display()))?;
let src_agents = agent_dir.join(".claude").join("agents");
let mut files = Vec::new();
for entry in
fs::read_dir(&src_agents).map_err(|e| format!("read {}: {e}", src_agents.display()))?
{
let entry = entry.map_err(|e| e.to_string())?;
let dst = agents_dir.join(entry.file_name());
fs::copy(entry.path(), &dst).map_err(|e| format!("copy agent: {e}"))?;
files.push(dst);
}
let settings_src = {
let split = agent_dir.join(".claude").join("settings.json");
if split.is_file() {
Some(split)
} else {
let flat = agent_dir.join("settings.json");
flat.is_file().then_some(flat)
}
};
if let Some(src) = settings_src {
let dst = claude_dir.join("settings.json");
fs::copy(&src, &dst).map_err(|e| format!("copy settings: {e}"))?;
files.push(dst);
}
Ok(InstalledAgents {
files,
agents_dir,
claude_dir,
})
}
pub(crate) fn agent_name(agent_dir: &Path) -> Result<String, String> {
let dir = agent_dir.join(".claude").join("agents");
let mut names: Vec<String> = fs::read_dir(&dir)
.map_err(|e| format!("read {}: {e}", dir.display()))?
.filter_map(|e| e.ok())
.filter_map(|e| e.file_name().to_str().map(str::to_string))
.filter_map(|f| f.strip_suffix(".md").map(str::to_string))
.collect();
names.sort();
names
.into_iter()
.next()
.ok_or_else(|| format!("no agent files in {}", dir.display()))
}
pub(crate) fn run_path(project: &Path) -> String {
let base = std::env::var("PATH").unwrap_or_default();
let venv = project.join(".venv").join("bin");
if venv.is_dir() {
format!("{}:{base}", venv.display())
} else {
base
}
}
fn run_case(
project: &Path,
agent: &str,
path_env: &str,
case: &GoldenCase,
sample: u32,
ctx: &CaseCtx,
) -> SampleOutcome {
let key = CaseKey {
case_id: &case.id,
question: &case.question,
agent_sha: ctx.agent_sha,
model: ctx.model,
context_sha: ctx.context_sha,
sample,
};
if let Some(trace) = ctx.store.load(&key) {
if let Some(verdict) = rescore(&trace, case) {
return SampleOutcome {
pass: verdict.pass,
reason: verdict.reason,
cost: trace.cost,
latency_ms: trace.latency_ms,
turns: trace.turns,
cache_hit: true,
answer: ctx.record_answers.then(|| render_answer(&trace.result)),
};
}
}
let started = Instant::now();
let fail = |reason: &str, cost: f64, latency_ms: u64| SampleOutcome {
pass: false,
reason: reason.to_string(),
cost,
latency_ms,
turns: 0,
cache_hit: false,
answer: None,
};
let mut args: Vec<&str> = vec!["-p", &case.question, "--agent", agent];
if let Some(model) = ctx.model_override() {
args.extend_from_slice(&["--model", model]);
}
args.extend_from_slice(&[
"--output-format",
"json",
"--allowedTools",
"Read",
"Bash(wren:*)",
]);
let output = Command::new("claude")
.args(&args)
.current_dir(project)
.env("PATH", path_env)
.output();
let raw = match output {
Ok(o) if o.status.success() => String::from_utf8_lossy(&o.stdout).into_owned(),
_ => {
return fail(
"claude invocation failed",
0.0,
started.elapsed().as_millis() as u64,
)
}
};
let meta: serde_json::Value = serde_json::from_str(&raw).unwrap_or(serde_json::Value::Null);
let result_text = meta.get("result").and_then(|v| v.as_str()).unwrap_or(&raw);
let latency_ms = meta
.get("duration_ms")
.and_then(|v| v.as_u64())
.unwrap_or_else(|| started.elapsed().as_millis() as u64);
let cost = meta
.get("total_cost_usd")
.and_then(|v| v.as_f64())
.unwrap_or(0.0);
let turns = meta.get("num_turns").and_then(|v| v.as_u64()).unwrap_or(0);
let extracted = match case.result_kind {
ResultKind::Table => extract_result_json(result_text),
ResultKind::Verdict => extract_verdict_json(result_text),
};
let Some(result_value) = extracted else {
let reason = match case.result_kind {
ResultKind::Table => "no parseable {columns,rows} in output",
ResultKind::Verdict => "no parseable verdict envelope in output",
};
return fail(reason, cost, latency_ms);
};
let Some(verdict) = score_value(&result_value, case) else {
let reason = match case.result_kind {
ResultKind::Table => "no parseable {columns,rows} in output",
ResultKind::Verdict => "verdict envelope missing the expected field",
};
return fail(reason, cost, latency_ms);
};
let trace = Trace {
case_id: case.id.clone(),
agent_sha: ctx.agent_sha.to_string(),
model: ctx.model.to_string(),
context_version: ctx.context_version.map(str::to_string),
context_sha: ctx.context_sha.to_string(),
question: case.question.clone(),
sql_executed: None,
result: result_value,
cost,
latency_ms,
turns,
tool_calls: None,
};
if let Err(e) = ctx.store.store(&key, &trace) {
eprintln!(
" note: could not cache trace for {} (sample {sample}): {e}",
case.id
);
}
SampleOutcome {
pass: verdict.pass,
reason: verdict.reason,
cost,
latency_ms,
turns,
cache_hit: false,
answer: ctx.record_answers.then(|| render_answer(&trace.result)),
}
}
pub fn run_eval(cfg: &RunConfig) -> Result<Report, String> {
let golden_text = fs::read_to_string(&cfg.golden_path)
.map_err(|e| format!("read {}: {e}", cfg.golden_path.display()))?;
let golden: Golden =
serde_yaml::from_str(&golden_text).map_err(|e| format!("parse golden: {e}"))?;
let total_cases = golden.cases.len();
let cases = select_and_subset(&golden, &cfg.filter)?;
let selected_cases = cases.len();
let agent = agent_name(&cfg.agent_dir)?;
let path_env = run_path(&cfg.project);
let (mut store, context_sha) =
build_store(cfg.no_cache, cfg.cache_dir.as_deref(), &cfg.project);
let agent_sha = if store.is_enabled() {
match context::compute_dir_sha(&cfg.agent_dir) {
Ok(s) => s,
Err(e) => {
eprintln!("cache: disabled — cannot hash agent dir: {e}");
store = TraceStore::disabled();
String::new()
}
}
} else {
String::new()
};
let _installed = install_agents(&cfg.agent_dir, &cfg.project)?;
let parallel = cfg.parallel.max(1);
let mut configs = Vec::new();
for model in &cfg.models {
let par = if parallel > 1 {
format!(", parallel={parallel}")
} else {
String::new()
};
eprintln!("\n### binding: strong→{model} (n={selected_cases}{par})");
let ctx = CaseCtx {
model,
agent_sha: &agent_sha,
context_sha: &context_sha,
context_version: golden.context_version.as_deref(),
store: &store,
record_answers: cfg.record_answers,
};
let rows = run_cases(
&cfg.project,
&agent,
&path_env,
&cases,
cfg.samples,
parallel,
&ctx,
);
configs.push(aggregate(model, rows));
}
Ok(Report {
dataset: golden.dataset,
context_version: golden.context_version,
parallel,
selected_cases,
total_cases,
configs,
})
}
pub(crate) fn build_store(
no_cache: bool,
cache_dir: Option<&Path>,
project: &Path,
) -> (TraceStore, String) {
if no_cache {
if cache_dir.is_some() {
eprintln!("cache: --cache-dir ignored — --no-cache disables the cache entirely");
}
eprintln!("cache: disabled (--no-cache) — every case re-runs and refreshes its trace");
return (TraceStore::disabled(), String::new());
}
let context_sha = match compute_mdl_sha(project) {
Ok(s) => s,
Err(e) => {
eprintln!("cache: disabled — cannot compute MDL context SHA: {e}");
return (TraceStore::disabled(), String::new());
}
};
let dir = cache_dir
.map(Path::to_path_buf)
.unwrap_or_else(|| project.join(".warble").join("eval-cache"));
eprintln!(
"cache: enabled at {} (context_sha {})",
dir.display(),
&context_sha[..context_sha.len().min(12)]
);
(TraceStore::new(dir, true), context_sha)
}
pub(crate) fn select_and_subset(
golden: &Golden,
filter: &CaseFilter,
) -> Result<Vec<GoldenCase>, String> {
let selection = select_cases(&golden.cases, filter);
if filter.is_active() {
eprintln!("{}", selection.note);
}
if filter.is_active() && selection.indices.is_empty() {
return Err(format!(
"no golden cases match the filter — {} (nothing to run)",
selection.note
));
}
Ok(selection
.indices
.iter()
.map(|&i| golden.cases[i].clone())
.collect())
}
pub(crate) fn run_cases(
project: &Path,
agent: &str,
path_env: &str,
cases: &[GoldenCase],
samples: usize,
parallel: usize,
ctx: &CaseCtx,
) -> Vec<CaseResult> {
let samples = samples.max(1);
let outcomes = run_indexed(cases.len() * samples, parallel, |j| {
let case = &cases[j / samples];
let sample = (j % samples) as u32;
let r = run_case(project, agent, path_env, case, sample, ctx);
let extra = if r.pass {
String::new()
} else {
format!(" — {}", r.reason)
};
let cost = if r.cost > 0.0 {
format!(", ${:.4}", r.cost)
} else {
String::new()
};
let cache = if r.cache_hit { " [cache]" } else { "" };
if samples == 1 {
eprintln!(
" {} {}{cache} ({:.1}s{cost}, {}t){extra}",
if r.pass { "PASS" } else { "FAIL" },
case.id,
r.latency_ms as f64 / 1000.0,
r.turns
);
} else {
eprintln!(
" {} {} sample {sample}{cache} ({:.1}s{cost}, {}t){extra}",
if r.pass { "PASS" } else { "FAIL" },
case.id,
r.latency_ms as f64 / 1000.0,
r.turns
);
}
r
});
let mut it = outcomes.into_iter();
cases
.iter()
.map(|case| {
let chunk: Vec<SampleOutcome> = (&mut it).take(samples).collect();
let result = fold_samples(
case.id.clone(),
case.tags.clone(),
chunk,
ctx.record_answers,
);
if samples > 1 {
let marker = if result.flaky {
"FLAKY"
} else if result.pass {
"PASS"
} else {
"FAIL"
};
eprintln!(
" {marker} {} pass_rate={:.2} ({}/{})",
result.id, result.pass_rate, result.passes, result.samples
);
}
result
})
.collect()
}
pub(crate) fn run_indexed<T, F>(n: usize, parallel: usize, job: F) -> Vec<T>
where
T: Send,
F: Fn(usize) -> T + Sync,
{
if parallel <= 1 || n <= 1 {
return (0..n).map(job).collect();
}
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Mutex;
let next = AtomicUsize::new(0);
let slots: Vec<Mutex<Option<T>>> = (0..n).map(|_| Mutex::new(None)).collect();
std::thread::scope(|scope| {
for _ in 0..parallel.min(n) {
scope.spawn(|| loop {
let i = next.fetch_add(1, Ordering::Relaxed);
if i >= n {
break;
}
let r = job(i);
*slots[i].lock().expect("result slot poisoned") = Some(r);
});
}
});
slots
.into_iter()
.map(|slot| {
slot.into_inner()
.expect("result slot poisoned")
.expect("every index 0..n is claimed exactly once by the shared counter")
})
.collect()
}
#[cfg(test)]
mod run_indexed_tests {
use super::run_indexed;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
#[test]
fn serial_path_preserves_order_and_runs_each_once() {
let calls = AtomicUsize::new(0);
let out = run_indexed(5, 1, |i| {
calls.fetch_add(1, Ordering::Relaxed);
i * 10
});
assert_eq!(out, vec![0, 10, 20, 30, 40]);
assert_eq!(calls.load(Ordering::Relaxed), 5);
}
#[test]
fn parallel_results_are_in_index_order_despite_completion_order() {
let out = run_indexed(8, 4, |i| {
std::thread::sleep(Duration::from_millis((8 - i as u64) * 3));
i
});
assert_eq!(out, (0..8).collect::<Vec<_>>());
}
#[test]
fn parallel_runs_each_index_exactly_once() {
let per_index: Vec<AtomicUsize> = (0..32).map(|_| AtomicUsize::new(0)).collect();
let out = run_indexed(32, 6, |i| {
per_index[i].fetch_add(1, Ordering::Relaxed);
i
});
assert_eq!(out.len(), 32);
for (i, c) in per_index.iter().enumerate() {
assert_eq!(c.load(Ordering::Relaxed), 1, "index {i} ran once");
}
}
#[test]
fn parallelism_wider_than_n_and_empty_input_are_fine() {
assert_eq!(run_indexed(2, 16, |i| i), vec![0, 1]);
assert_eq!(run_indexed(0, 4, |i| i), Vec::<usize>::new());
}
#[test]
fn parallel_actually_overlaps_work() {
let in_flight = AtomicUsize::new(0);
let max_in_flight = AtomicUsize::new(0);
run_indexed(4, 4, |_| {
let now = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
max_in_flight.fetch_max(now, Ordering::SeqCst);
std::thread::sleep(Duration::from_millis(40));
in_flight.fetch_sub(1, Ordering::SeqCst);
});
let max = max_in_flight.load(Ordering::SeqCst);
assert!(
max > 1,
"expected overlapping execution, but max concurrent was {max}"
);
}
}
#[cfg(test)]
mod fold_samples_tests {
use super::*;
fn outcome(pass: bool, reason: &str, cost: f64, latency_ms: u64, turns: u64) -> SampleOutcome {
SampleOutcome {
pass,
reason: reason.to_string(),
cost,
latency_ms,
turns,
cache_hit: false,
answer: None,
}
}
#[test]
fn fully_passing_is_not_flaky_and_reason_is_match() {
let outcomes = vec![
outcome(true, "match", 0.1, 1_000, 2),
outcome(true, "match", 0.1, 1_000, 2),
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert_eq!(r.samples, 2);
assert_eq!(r.passes, 2);
assert!((r.pass_rate - 1.0).abs() < 1e-9);
assert!(r.pass);
assert!(!r.flaky);
assert_eq!(r.reason, "match");
}
#[test]
fn fully_failing_is_not_flaky() {
let outcomes = vec![
outcome(false, "mismatch", 0.0, 500, 1),
outcome(false, "mismatch", 0.0, 500, 1),
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert_eq!(r.passes, 0);
assert_eq!(r.pass_rate, 0.0);
assert!(!r.pass);
assert!(!r.flaky, "0 passes out of N is a hard fail, not flaky");
}
#[test]
fn partial_pass_is_flaky_and_reason_is_the_modal_failure() {
let outcomes = vec![
outcome(true, "match", 0.0, 0, 0),
outcome(false, "wrong total", 0.0, 0, 0),
outcome(false, "wrong total", 0.0, 0, 0),
outcome(false, "timeout", 0.0, 0, 0),
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert_eq!(r.passes, 1);
assert_eq!(r.samples, 4);
assert!((r.pass_rate - 0.25).abs() < 1e-9);
assert!(!r.pass);
assert!(r.flaky);
assert_eq!(r.reason, "wrong total", "modal failure reason wins");
}
#[test]
fn modal_reason_tie_break_is_first_occurrence() {
let outcomes = vec![
outcome(false, "a", 0.0, 0, 0),
outcome(false, "b", 0.0, 0, 0),
outcome(false, "b", 0.0, 0, 0),
outcome(false, "a", 0.0, 0, 0),
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert_eq!(r.reason, "a");
}
#[test]
fn cost_sums_latency_and_turns_average() {
let outcomes = vec![
outcome(true, "match", 0.10, 1_000, 4),
outcome(true, "match", 0.20, 3_000, 6),
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert!((r.cost - 0.30).abs() < 1e-9, "cost sums across samples");
assert_eq!(r.latency_ms, 2_000, "latency averages across samples");
assert_eq!(r.turns, 5, "turns averages across samples");
}
#[test]
fn cache_hits_and_misses_are_counted_per_sample() {
let outcomes = vec![
SampleOutcome {
cache_hit: true,
..outcome(true, "match", 0.0, 0, 0)
},
SampleOutcome {
cache_hit: false,
..outcome(true, "match", 0.0, 0, 0)
},
SampleOutcome {
cache_hit: true,
..outcome(true, "match", 0.0, 0, 0)
},
];
let r = fold_samples("a".into(), vec![], outcomes, false);
assert_eq!(r.cache_hits, 2);
assert_eq!(r.cache_misses, 1);
}
#[test]
fn answer_dist_is_only_populated_when_record_answers_is_set() {
let build = || {
vec![
SampleOutcome {
answer: Some("42".to_string()),
..outcome(true, "match", 0.0, 0, 0)
},
SampleOutcome {
answer: Some("42".to_string()),
..outcome(true, "match", 0.0, 0, 0)
},
SampleOutcome {
answer: Some("7".to_string()),
..outcome(false, "wrong total", 0.0, 0, 0)
},
]
};
let without = fold_samples("a".into(), vec![], build(), false);
assert!(without.answer_dist.is_none());
assert!(
without.samples_detail.iter().all(|s| s.answer.is_none()),
"no answer strings recorded unless --record-answers"
);
let with = fold_samples("a".into(), vec![], build(), true);
let dist = with.answer_dist.expect("populated when record_answers");
assert_eq!(dist.get("42"), Some(&2));
assert_eq!(dist.get("7"), Some(&1));
assert_eq!(
with.samples_detail
.iter()
.filter(|s| s.answer.is_some())
.count(),
3
);
}
#[test]
fn modal_reason_of_empty_iterator_is_empty_string() {
assert_eq!(modal_reason(std::iter::empty()), "");
}
}
#[cfg(test)]
mod verdict_tests {
use super::*;
fn envelope(fresh: bool, severity: &str) -> serde_json::Value {
serde_json::json!({
"blocks": [
{ "type": "status", "state": if fresh { "fresh" } else { "stale" },
"label": "orders freshness", "detail": "…", "severity": severity }
],
"verdict": { "type": "freshness_verdict", "fresh": fresh, "observed_lag_hours": 51,
"expected_cadence": "24h" },
"emitted": ["freshness_breach"],
"verified": true,
})
}
fn fenced(value: &serde_json::Value) -> String {
format!("Some prose.\n\n```json\n{value}\n```\n")
}
#[test]
fn extract_verdict_json_accepts_a_real_envelope() {
let text = fenced(&envelope(false, "critical"));
let got = extract_verdict_json(&text).expect("envelope parses");
assert_eq!(got["verdict"]["fresh"], false);
}
#[test]
fn extract_verdict_json_rejects_a_columns_rows_table() {
let text = fenced(&serde_json::json!({"columns": ["n"], "rows": [[42]]}));
assert!(
extract_verdict_json(&text).is_none(),
"a table has neither `verdict` nor `blocks`"
);
}
#[test]
fn extract_verdict_json_rejects_malformed_json() {
assert!(extract_verdict_json("not json at all").is_none());
assert!(extract_verdict_json("```json\n{ not valid\n```").is_none());
}
#[test]
fn extract_result_json_rejects_a_verdict_envelope() {
let text = fenced(&envelope(true, "critical"));
assert!(extract_result_json(&text).is_none());
}
#[test]
fn project_verdict_field_reads_fresh_true_and_false() {
let fresh_table = project_verdict_field(&envelope(true, "critical"), "fresh").unwrap();
assert_eq!(fresh_table.rows, vec![vec![Value::Bool(true)]]);
let stale_table = project_verdict_field(&envelope(false, "critical"), "fresh").unwrap();
assert_eq!(stale_table.rows, vec![vec![Value::Bool(false)]]);
}
#[test]
fn project_verdict_field_reads_severity_warn_and_critical() {
let warn = project_verdict_field(&envelope(false, "warn"), "severity").unwrap();
assert_eq!(warn.rows, vec![vec![Value::String("warn".to_string())]]);
let critical = project_verdict_field(&envelope(false, "critical"), "severity").unwrap();
assert_eq!(
critical.rows,
vec![vec![Value::String("critical".to_string())]]
);
}
#[test]
fn project_verdict_field_falls_back_to_verdict_severity() {
let envelope = serde_json::json!({
"blocks": [{ "type": "status", "state": "stale" }],
"verdict": { "type": "freshness_verdict", "fresh": false, "severity": "warn" },
});
let got = project_verdict_field(&envelope, "severity").unwrap();
assert_eq!(got.rows, vec![vec![Value::String("warn".to_string())]]);
}
#[test]
fn project_verdict_field_is_none_on_missing_or_unknown_field() {
let e = envelope(true, "critical");
assert!(
project_verdict_field(&e, "observed_lag_hours").is_none(),
"unrecognized field fails closed, never silently passes"
);
let no_verdict = serde_json::json!({"blocks": []});
assert!(project_verdict_field(&no_verdict, "fresh").is_none());
let no_severity_anywhere = serde_json::json!({
"blocks": [{"type": "status", "state": "stale"}],
"verdict": {"type": "freshness_verdict", "fresh": false},
});
assert!(project_verdict_field(&no_severity_anywhere, "severity").is_none());
}
fn verdict_case(tag: &str, field: &str, expected_rows: &str) -> GoldenCase {
let yaml = format!(
"id: v1\nquestion: \"is it fresh?\"\ntags: [{tag}]\nmatch: scalar\n\
result_kind: verdict\nverdict_field: {field}\n\
expected: {{ columns: [{field}], rows: {expected_rows} }}\n"
);
serde_yaml::from_str(&yaml).expect("verdict golden case parses")
}
#[test]
fn golden_case_result_kind_defaults_to_table_when_omitted() {
let yaml = "id: t1\nquestion: \"how many?\"\nmatch: scalar\nexpected: { columns: [n], rows: [[1]] }\n";
let case: GoldenCase = serde_yaml::from_str(yaml).expect("legacy case parses");
assert_eq!(case.result_kind, ResultKind::Table);
assert!(case.verdict_field.is_none());
}
#[test]
fn score_value_passes_a_matching_verdict_envelope() {
let case = verdict_case("detection", "fresh", "[[false]]");
let verdict = score_value(&envelope(false, "critical"), &case).expect("scores");
assert!(
verdict.pass,
"envelope fresh=false matches expected fresh=false"
);
}
#[test]
fn score_value_fails_a_mismatching_verdict_envelope() {
let case = verdict_case("detection", "fresh", "[[true]]");
let verdict = score_value(&envelope(false, "critical"), &case).expect("scores");
assert!(
!verdict.pass,
"envelope fresh=false does not match expected fresh=true"
);
}
#[test]
fn score_value_projects_severity_for_the_severity_field() {
let case = verdict_case("severity", "severity", "[[critical]]");
let verdict = score_value(&envelope(false, "critical"), &case).expect("scores");
assert!(verdict.pass);
}
#[test]
fn score_value_on_a_table_case_is_unaffected_by_verdict_support() {
let yaml = "id: t1\nquestion: \"how many?\"\nmatch: scalar\nexpected: { columns: [n], rows: [[42]] }\n";
let case: GoldenCase = serde_yaml::from_str(yaml).expect("legacy case parses");
let result = serde_json::json!({"columns": ["n"], "rows": [[42]]});
let verdict = score_value(&result, &case).expect("table case scores");
assert!(verdict.pass);
}
}