From 963fa5da3aee4d8f9a646affa7049009a57fd1c4 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 01:29:37 +0800 Subject: [PATCH] Isolate evaluation scheduling from recording and reporting --- src/eval/error.rs | 1 + src/eval/executor.rs | 475 +++++++++++++++++++++++++++++++++++++++++++ src/eval/mod.rs | 3 + src/eval/recorder.rs | 164 +++++++++++++++ src/eval/reporter.rs | 44 ++++ src/eval/runner.rs | 262 ++---------------------- src/eval/types.rs | 4 +- tests/eval_test.rs | 90 +++++++- 8 files changed, 789 insertions(+), 254 deletions(-) create mode 100644 src/eval/executor.rs create mode 100644 src/eval/recorder.rs create mode 100644 src/eval/reporter.rs diff --git a/src/eval/error.rs b/src/eval/error.rs index 9798a2f..6d9dc92 100644 --- a/src/eval/error.rs +++ b/src/eval/error.rs @@ -75,6 +75,7 @@ pub enum EvalRunFailure { /// An incomplete run preserves finished case results and every run-level failure. /// Case computation errors remain in those results rather than aborting the run. +/// Recording failures do not discard computed results or change their success counts. pub struct EvalRunError { pub failures: Vec, pub partial: Option>>, diff --git a/src/eval/executor.rs b/src/eval/executor.rs new file mode 100644 index 0000000..372939c --- /dev/null +++ b/src/eval/executor.rs @@ -0,0 +1,475 @@ +//! Bounded run scheduling and aggregation, independent of recording and output. +use async_trait::async_trait; +use futures::{ + future::{select, Either}, + stream::{FuturesUnordered, StreamExt}, +}; + +use super::{ + case_runner::CaseRunner, Case, EvalDefinition, EvalOpts, EvalResult, EvalRunError, + EvalRunFailure, EvalSummary, +}; +use crate::{BraintrustError, Result, SpanHandle}; + +/// An invocation owns its source and options; only the definition is shared. +pub(crate) struct EvalPlan { + name: String, + source: Box>, + definition: EvalDefinition, + concurrency: usize, +} +impl EvalPlan { + pub fn new(opts: EvalOpts) -> std::result::Result { + opts.validate()?; + Ok(Self { + name: opts.name, + source: opts.source, + definition: opts.definition, + concurrency: opts.max_concurrency, + }) + } +} + +#[async_trait] +pub(crate) trait EvalRecorder: Send + Sync { + async fn start(&self, index: usize, case: &Case) -> Result>>; + async fn finish(&self) -> Vec; +} + +#[async_trait] +pub(crate) trait CaseRecording: Send { + fn span(&self) -> Option<&SpanHandle>; + async fn finish(self: Box, result: &mut EvalResult) -> Vec; +} + +pub(crate) trait EvalReporter { + fn started(&mut self, name: &str); + fn case_finished(&mut self, index: usize, result: &EvalResult); + fn finished(&mut self, summary: &EvalSummary); +} + +struct CaseOutcome { + result: EvalResult, + failures: Vec, +} + +pub(crate) struct EvalExecutor; +impl EvalExecutor { + pub async fn run( + mut plan: EvalPlan, + recorder: Option<&dyn EvalRecorder>, + reporter: &mut impl EvalReporter, + ) -> std::result::Result, EvalRunError> { + reporter.started(&plan.name); + let dataset = plan.source.provenance(); + let mut results = Vec::new(); + let mut failures = Vec::new(); + let mut case_count = 0; + let mut pending = FuturesUnordered::new(); + loop { + // Keep one source read alive while cases run. Sources may await + // case progress and need not support cancellation/restarting next(). + let mut next = Box::pin(plan.source.next()); + let next_case = loop { + if pending.is_empty() { + break next.await; + } + match select(next.as_mut(), pending.next()).await { + Either::Left((case, _)) => break case, + Either::Right((Some((index, outcome)), _)) => { + Self::collect(index, outcome, &mut results, &mut failures, reporter); + if !failures.is_empty() { + break Ok(None); + } + } + Either::Right((None, _)) => unreachable!("nonempty pending cases"), + } + }; + let case = match next_case { + Ok(Some(case)) => case, + Ok(None) => break, + Err(source) => { + failures.push(EvalRunFailure::Source { + case_index: case_count, + source, + }); + break; + } + }; + let index = case_count; + case_count += 1; + let definition = &plan.definition; + pending.push(async move { + ( + index, + Self::run_case(case, definition, recorder, index).await, + ) + }); + if pending.len() >= plan.concurrency { + if let Some((index, outcome)) = pending.next().await { + Self::collect(index, outcome, &mut results, &mut failures, reporter); + } + if !failures.is_empty() { + break; + } + } + } + // Stop reading after failure, but finish all cases already accepted. + while let Some((index, outcome)) = pending.next().await { + Self::collect(index, outcome, &mut results, &mut failures, reporter); + } + if let Some(recorder) = recorder { + failures.extend(recorder.finish().await.into_iter().map(|source| { + EvalRunFailure::Recording { + case_index: None, + source, + } + })); + } + results.sort_unstable_by_key(|(index, _)| *index); + let results: Vec<_> = results.into_iter().map(|(_, result)| result).collect(); + let score_stats = EvalSummary::::compute_stats(&results); + let successful_cases = results + .iter() + .filter(|result| result.errors.is_empty()) + .count(); + let summary = EvalSummary { + name: plan.name, + results, + score_stats, + total_cases: case_count, + successful_cases, + failed_cases: case_count - successful_cases, + dataset, + }; + reporter.finished(&summary); + if failures.is_empty() { + Ok(summary) + } else { + Err(EvalRunError { + failures, + partial: Some(Box::new(summary)), + }) + } + } + + async fn run_case( + case: Case, + definition: &EvalDefinition, + recorder: Option<&dyn EvalRecorder>, + index: usize, + ) -> CaseOutcome { + let mut failures = Vec::new(); + let recording = if let Some(recorder) = recorder { + match recorder.start(index, &case).await { + Ok(recording) => Some(recording), + Err(error) => { + failures.push(error); + None + } + } + } else { + None + }; + let span = recording.as_ref().and_then(|recording| recording.span()); + let mut result = CaseRunner::run(case, definition.task(), definition.scorers(), span).await; + if let Some(recording) = recording { + failures.extend(recording.finish(&mut result).await); + } + CaseOutcome { result, failures } + } + + fn collect( + index: usize, + outcome: CaseOutcome, + results: &mut Vec<(usize, EvalResult)>, + failures: &mut Vec, + reporter: &mut impl EvalReporter, + ) { + failures.extend( + outcome + .failures + .into_iter() + .map(|source| EvalRunFailure::Recording { + case_index: Some(index), + source, + }), + ); + reporter.case_finished(index, &outcome.result); + results.push((index, outcome.result)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::eval::{task, CaseSource, SimpleFnTask, SourceError}; + use std::{ + cell::Cell, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + }; + + struct Input(usize); + struct Output(usize); + fn case(input: usize) -> Case { + Case { + input: Input(input), + expected: None, + metadata: None, + tags: None, + id: Some(input.to_string()), + xact_id: None, + created: None, + } + } + #[derive(Default)] + struct Observed { + cases: Vec, + finished: bool, + } + impl EvalReporter for Observed { + fn started(&mut self, _: &str) {} + fn case_finished(&mut self, index: usize, _: &EvalResult) { + self.cases.push(index); + } + fn finished(&mut self, _: &EvalSummary) { + self.finished = true; + } + } + struct WaitingSource { + cursor: Cell, + task_progress: Option>, + reads: Arc, + } + #[async_trait] + impl CaseSource for WaitingSource { + async fn next(&mut self) -> std::result::Result>, SourceError> { + self.reads.fetch_add(1, Ordering::SeqCst); + let index = self.cursor.get(); + self.cursor.set(index + 1); + match index { + 0 => Ok(Some(case(1))), + 1 => { + self.task_progress.take().unwrap().await.unwrap(); + Ok(Some(case(2))) + } + _ => anyhow::bail!("source stopped"), + } + } + } + + #[test] + fn local_run_drives_pending_source_and_drains_cases_without_runtime_or_serialization() { + let (progress, receiver) = futures::channel::oneshot::channel(); + let progress = Mutex::new(Some(progress)); + let reads = Arc::new(AtomicUsize::new(0)); + let definition = EvalDefinition::new( + Box::new(task(move |input: &Input, _hooks| { + if let Some(progress) = progress.lock().unwrap().take() { + progress.send(()).unwrap(); + } + Box::pin(async move { Ok(Output(input.0 * 2)) }) + })), + vec![], + ) + .unwrap(); + let plan = EvalPlan::new( + EvalOpts::builder() + .name("local".into()) + .source(Box::new(WaitingSource { + cursor: Cell::new(0), + task_progress: Some(receiver), + reads: reads.clone(), + })) + .definition(definition) + .max_concurrency(2) + .build(), + ) + .unwrap(); + let mut reporter = Observed::default(); + let error = futures::executor::block_on(EvalExecutor::run(plan, None, &mut reporter)) + .err() + .expect("source should fail"); + assert!( + matches!(&error.failures[..], [EvalRunFailure::Source {case_index: 2, source}] if source.to_string() == "source stopped") + ); + let summary = error.partial.unwrap(); + assert_eq!( + reads.load(Ordering::SeqCst), + 3, + "pending next() must not restart" + ); + assert_eq!(summary.total_cases, 2); + assert_eq!(summary.successful_cases, 2); + assert_eq!( + summary + .results + .iter() + .map(|result| result.output.as_ref().unwrap().0) + .collect::>(), + [2, 4] + ); + assert!(summary + .results + .iter() + .all(|result| result.span_id.is_none())); + assert_eq!(reporter.cases, [0, 1]); + assert!(reporter.finished); + } + + #[test] + fn local_success_has_bounded_concurrency_and_source_order_without_data_trait_bounds() { + let active = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let barrier = Arc::new(tokio::sync::Barrier::new(2)); + let definition = EvalDefinition::new( + Box::new(task({ + let active = active.clone(); + let peak = peak.clone(); + move |input: &Input, _hooks| { + let active = active.clone(); + let peak = peak.clone(); + let barrier = barrier.clone(); + Box::pin(async move { + let count = active.fetch_add(1, Ordering::SeqCst) + 1; + peak.fetch_max(count, Ordering::SeqCst); + barrier.wait().await; + active.fetch_sub(1, Ordering::SeqCst); + Ok(Output(input.0 * 2)) + }) + } + })), + vec![], + ) + .unwrap(); + for count in [4, 0] { + let plan = EvalPlan::new( + EvalOpts::builder() + .name("local-success".into()) + .source((0..count).map(case).collect::>().into()) + .definition(definition.clone()) + .max_concurrency(2) + .build(), + ) + .unwrap(); + let mut reporter = Observed::default(); + let summary = futures::executor::block_on(EvalExecutor::run(plan, None, &mut reporter)) + .expect("local computation should succeed"); + assert_eq!(summary.total_cases, count); + assert_eq!(summary.successful_cases, count); + assert_eq!( + summary + .results + .iter() + .map(|result| result.output.as_ref().unwrap().0) + .collect::>(), + (0..count).map(|index| index * 2).collect::>() + ); + assert!(reporter.finished); + } + assert_eq!(active.load(Ordering::SeqCst), 0); + assert_eq!(peak.load(Ordering::SeqCst), 2); + } + + struct FailingRecorder { + finished: Arc, + } + struct FailingCase; + #[async_trait] + impl EvalRecorder for FailingRecorder { + async fn start( + &self, + index: usize, + _: &Case, + ) -> Result>> { + if index == 0 { + return Err(BraintrustError::Background("start failed".into())); + } + Ok(Box::new(FailingCase)) + } + async fn finish(&self) -> Vec { + self.finished.fetch_add(1, Ordering::SeqCst); + vec![BraintrustError::Background("run finish failed".into())] + } + } + #[async_trait] + impl CaseRecording for FailingCase { + fn span(&self) -> Option<&SpanHandle> { + None + } + async fn finish(self: Box, _: &mut EvalResult) -> Vec { + vec![BraintrustError::Background("case finish failed".into())] + } + } + #[test] + fn recorder_start_and_finish_errors_preserve_computation_and_finalize_run() { + let finished = Arc::new(AtomicUsize::new(0)); + let recorder = FailingRecorder { + finished: finished.clone(), + }; + let cases = (0..3) + .map(|input| Case { + input, + ..Default::default() + }) + .collect::>>(); + let plan = EvalPlan::new( + EvalOpts::builder() + .name("recorded".into()) + .source(cases.into()) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &usize| Ok(input + 10))), + vec![], + ) + .unwrap(), + ) + .max_concurrency(2) + .build(), + ) + .unwrap(); + let mut reporter = Observed::default(); + let error = + futures::executor::block_on(EvalExecutor::run(plan, Some(&recorder), &mut reporter)) + .unwrap_err(); + assert_eq!(error.failures.len(), 3); + assert!(matches!( + &error.failures[0], + EvalRunFailure::Recording { + case_index: Some(0), + .. + } + )); + assert!(matches!( + &error.failures[1], + EvalRunFailure::Recording { + case_index: Some(1), + .. + } + )); + assert!(matches!( + &error.failures[2], + EvalRunFailure::Recording { + case_index: None, + .. + } + )); + let summary = error.partial.unwrap(); + assert_eq!( + summary + .results + .iter() + .map(|result| result.output) + .collect::>(), + [Some(10), Some(11)] + ); + assert_eq!(summary.total_cases, 2, "third case must remain unread"); + assert_eq!(summary.successful_cases, 2); + assert_eq!(summary.failed_cases, 0); + assert_eq!(finished.load(Ordering::SeqCst), 1); + assert!(reporter.finished); + } +} diff --git a/src/eval/mod.rs b/src/eval/mod.rs index 22d02cc..8fe357f 100644 --- a/src/eval/mod.rs +++ b/src/eval/mod.rs @@ -55,6 +55,9 @@ mod bt_runner; mod case_runner; mod definition; mod error; +mod executor; +mod recorder; +mod reporter; mod runner; mod scorer; mod source; diff --git a/src/eval/recorder.rs b/src/eval/recorder.rs new file mode 100644 index 0000000..0f8847f --- /dev/null +++ b/src/eval/recorder.rs @@ -0,0 +1,164 @@ +//! Braintrust serialization and write completions, outside the run engine. +use super::{ + executor::{CaseRecording, EvalRecorder}, + Case, EvalResult, +}; +use crate::{ + BraintrustClient, BraintrustError, Completion, DeliveryReceipt, Result, SpanHandle, SpanLog, + WriteError, +}; +use async_trait::async_trait; +use serde::Serialize; +use serde_json::{Map, Value}; +use std::sync::Arc; + +type Write = Completion; + +pub(crate) struct BraintrustRecorder { + client: Arc, + root: SpanHandle, + root_write: tokio::sync::Mutex>, +} +impl BraintrustRecorder { + pub async fn new( + client: Arc, + name: String, + metadata: Option>, + tags: Option>, + ) -> Result { + let root = client + .span_builder() + .await? + .span_type(crate::SpanType::Eval) + .build(); + let mut log = SpanLog::builder().name(format!("eval-{name}")); + if let Some(metadata) = metadata { + log = log.metadata(metadata); + } + if let Some(tags) = tags { + log = log.tags(tags); + } + let root_write = tokio::sync::Mutex::new(Some(root.log(log.build().unwrap()))); + Ok(Self { + client, + root, + root_write, + }) + } +} + +struct BraintrustCase { + span: SpanHandle, + initial_write: Write, + failures: Vec, +} +fn serialize( + value: &T, + field: &str, + failures: &mut Vec, +) -> Option { + match serde_json::to_value(value) { + Ok(value) => Some(value), + Err(error) => { + failures.push(BraintrustError::Background(format!( + "failed to serialize evaluation {field}: {error}" + ))); + None + } + } +} +#[async_trait] +impl EvalRecorder + for BraintrustRecorder +{ + async fn start(&self, index: usize, case: &Case) -> Result>> { + let span = self + .client + .span_builder() + .await? + .span_type(crate::SpanType::Eval) + .build(); + let mut failures = Vec::new(); + let mut log = SpanLog::builder().name(format!("case-{}", index + 1)); + if let Some(input) = serialize(&case.input, "input", &mut failures) { + log = log.input(input); + } + if let Some(expected) = &case.expected { + if let Some(expected) = serialize(expected, "expected", &mut failures) { + log = log.expected(expected); + } + } + if let Some(metadata) = &case.metadata { + log = log.metadata(metadata.clone()); + } + if let Some(tags) = &case.tags { + log = log.tags(tags.clone()); + } + let initial_write = span.log(log.build().unwrap()); + Ok(Box::new(BraintrustCase { + span, + initial_write, + failures, + })) + } + async fn finish(&self) -> Vec { + let end = self.root.end(); + let mut failures = Vec::new(); + if let Some(write) = self.root_write.lock().await.take() { + if let Err(error) = write.await { + failures.push(error.into()); + } + } + if let Err(error) = end.await { + failures.push(error.into()); + } + failures + } +} +#[async_trait] +impl CaseRecording + for BraintrustCase +{ + fn span(&self) -> Option<&SpanHandle> { + Some(&self.span) + } + async fn finish(mut self: Box, result: &mut EvalResult) -> Vec { + match self.span.export() { + Ok(export) => result.span_id = export.span_id, + Err(error) => self.failures.push(error), + } + let mut log = SpanLog::builder().scores( + result + .scores + .iter() + .map(|score| (score.name.clone(), score.score)) + .collect(), + ); + if let Some(output) = &result.output { + if let Some(output) = serialize(output, "output", &mut self.failures) { + log = log.output(output); + } + } + if let Some(metadata) = &result.metadata { + log = log.metadata(metadata.clone()); + } + if let Some(tags) = &result.tags { + log = log.tags(tags.clone()); + } + if !result.errors.is_empty() { + if let Some(errors) = serialize(&result.errors, "case errors", &mut self.failures) { + log = log.error(errors); + } + } + let final_write = self.span.log(log.build().unwrap()); + let end = self.span.end(); + // Observe every completion, even if serialization or an earlier write + // failed. The owned computation result is never discarded here. + for outcome in [self.initial_write.await, final_write.await, end.await] { + if let Err(error) = outcome { + self.failures.push(error.into()); + } + } + self.failures + } +} diff --git a/src/eval/reporter.rs b/src/eval/reporter.rs new file mode 100644 index 0000000..149fce5 --- /dev/null +++ b/src/eval/reporter.rs @@ -0,0 +1,44 @@ +//! Console presentation, separate from execution and recording. +use super::{executor::EvalReporter, EvalResult, EvalSummary}; + +pub(crate) struct TerminalReporter { + pub quiet: bool, +} +impl EvalReporter for TerminalReporter { + fn started(&mut self, name: &str) { + if !self.quiet { + println!("Running evaluation: {name}"); + } + } + fn case_finished(&mut self, index: usize, result: &EvalResult) { + if self.quiet { + return; + } + let scores = result + .scores + .iter() + .map(|score| format!("{}={:.3}", score.name, score.score)) + .collect::>() + .join(", "); + println!("Case {}: {}", index + 1, scores); + for error in &result.errors { + eprintln!("Case {}: {}", index + 1, error); + } + } + fn finished(&mut self, summary: &EvalSummary) { + if self.quiet { + return; + } + println!("\nEvaluation complete!"); + println!("Total cases: {}", summary.total_cases); + println!("Successful: {}", summary.successful_cases); + println!("Failed: {}", summary.failed_cases); + println!("\nScore summary:"); + for stat in &summary.score_stats { + println!( + " {}: mean={:.3}, min={:.3}, max={:.3}, count={}", + stat.name, stat.mean, stat.min, stat.max, stat.count + ); + } + } +} diff --git a/src/eval/runner.rs b/src/eval/runner.rs index 8263a28..498b11e 100644 --- a/src/eval/runner.rs +++ b/src/eval/runner.rs @@ -1,16 +1,12 @@ -use futures::stream::{FuturesUnordered, StreamExt}; use serde::Serialize; use serde_json::{Map, Value}; use std::marker::PhantomData; use std::sync::Arc; -use crate::{BraintrustClient, Result, SpanHandle}; +use crate::{BraintrustClient, Result}; use super::{ - case_runner::CaseRunner, - source::CaseSource, - types::{Case, EvalResult, EvalSummary}, - EvalConfigError, EvalDefinition, EvalRunError, EvalRunFailure, + source::CaseSource, types::EvalSummary, EvalConfigError, EvalDefinition, EvalRunError, }; /// Options for running an evaluation @@ -78,251 +74,17 @@ where /// Run the evaluation pub async fn run( &self, - mut opts: EvalOpts, + opts: EvalOpts, ) -> std::result::Result, EvalRunError> { - opts.validate()?; - if !opts.quiet { - println!("Running evaluation: {}", opts.name); - } - - // Create root span for the evaluation - let root_span = self - .client - .span_builder() - .await? - .span_type(crate::SpanType::Eval) - .build(); - - // Set name and metadata on the span - let mut log_builder = crate::SpanLog::builder().name(format!("eval-{}", opts.name)); - - if let Some(metadata) = &opts.metadata { - log_builder = log_builder.metadata(metadata.clone()); - } - if let Some(tags) = &opts.tags { - log_builder = log_builder.tags(tags.clone()); - } - - let root_write = root_span.log(log_builder.build().unwrap()); - - let dataset = opts.source.provenance(); - let mut results = Vec::new(); - let mut failures = Vec::new(); - let mut case_count = 0; - let mut futures = FuturesUnordered::new(); - - loop { - // Retain the next-read future while polling active cases. A source - // may wait for prior computation, and need not be cancellation-safe. - let mut next = Box::pin(opts.source.next()); - let next_case = loop { - if futures.is_empty() { - break next.await; - } - tokio::select! { - next_case = next.as_mut() => break next_case, - Some((index, result)) = futures.next() => { - Self::collect_case_result(index, result, &mut results, &mut failures); - if !failures.is_empty() { break Ok(None); } - } - } - }; - let case = match next_case { - Ok(Some(case)) => case, - Ok(None) => break, - Err(source) => { - failures.push(EvalRunFailure::Source { - case_index: case_count, - source, - }); - break; - } - }; - case_count += 1; - let index = case_count - 1; - let client = Arc::clone(&self.client); - let root_span = root_span.clone(); - let definition = &opts.definition; - let quiet = opts.quiet; - futures.push(async move { - let result = - Self::process_case(case, definition, &root_span, &client, index + 1, quiet) - .await; - (index, result) - }); - if futures.len() >= opts.max_concurrency { - if let Some((index, result)) = futures.next().await { - Self::collect_case_result(index, result, &mut results, &mut failures); - } - if !failures.is_empty() { - break; - } - } - } - // Source/recording failure does not abandon already scheduled cases. - while let Some((index, result)) = futures.next().await { - Self::collect_case_result(index, result, &mut results, &mut failures); - } - let end = root_span.end(); - for result in [root_write.await, end.await] { - if let Err(error) = result { - failures.push(EvalRunFailure::Recording { - case_index: None, - source: error.into(), - }); - } - } - results.sort_unstable_by_key(|(index, _)| *index); - let results: Vec<_> = results.into_iter().map(|(_, result)| result).collect(); - - // Compute statistics - let score_stats = EvalSummary::::compute_stats(&results); - let successful = results.iter().filter(|r| r.errors.is_empty()).count(); - let failed = case_count - successful; - - if !opts.quiet { - println!("\nEvaluation complete!"); - println!("Total cases: {}", results.len()); - println!("Successful: {}", successful); - println!("Failed: {}", failed); - println!("\nScore summary:"); - for stat in &score_stats { - println!( - " {}: mean={:.3}, min={:.3}, max={:.3}, count={}", - stat.name, stat.mean, stat.min, stat.max, stat.count - ); - } - } - - let summary = EvalSummary { - name: opts.name, - results, - score_stats, - total_cases: case_count, - successful_cases: successful, - failed_cases: failed, - dataset, - }; - if failures.is_empty() { - Ok(summary) - } else { - Err(EvalRunError { - failures, - partial: Some(Box::new(summary)), - }) - } - } - - fn collect_case_result( - index: usize, - result: Result>, - results: &mut Vec<(usize, EvalResult)>, - failures: &mut Vec, - ) { - match result { - Ok(result) => results.push((index, result)), - Err(source) => failures.push(EvalRunFailure::Recording { - case_index: Some(index), - source, - }), - } - } - - /// Process a single case - async fn process_case( - case: Case, - definition: &EvalDefinition, - _root_span: &SpanHandle, - client: &BraintrustClient, - case_num: usize, - quiet: bool, - ) -> Result> { - let case_span = client - .span_builder() - .await? - .span_type(crate::SpanType::Eval) - .build(); - - // Build initial span log with name, input, and expected - let mut log_builder = crate::SpanLog::builder().name(format!("case-{}", case_num)); - - if let Ok(input_json) = serde_json::to_value(&case.input) { - log_builder = log_builder.input(input_json); - } - - if let Some(expected) = &case.expected { - if let Ok(expected_json) = serde_json::to_value(expected) { - log_builder = log_builder.expected(expected_json); - } - } - - // Add case metadata and tags - if let Some(metadata) = &case.metadata { - log_builder = log_builder.metadata(metadata.clone()); - } - if let Some(tags) = &case.tags { - log_builder = log_builder.tags(tags.clone()); - } - - let initial_write = case_span.log(log_builder.build().unwrap()); - let mut result = CaseRunner::run( - case, - definition.task(), - definition.scorers(), - Some(&case_span), - ) - .await; - result.span_id = case_span.export()?.span_id; - - // Recording adapts the completed outcome; computation never waits on - // HTTP or emits console messages. Observe every accepted write before - // reporting recording success, including writes that failed earlier. - let mut final_log = crate::SpanLog::builder().scores( - result - .scores - .iter() - .map(|score| (score.name.clone(), score.score)) - .collect(), - ); - if let Some(output) = &result.output { - if let Ok(output) = serde_json::to_value(output) { - final_log = final_log.output(output); - } - } - if let Some(metadata) = &result.metadata { - final_log = final_log.metadata(metadata.clone()); - } - if let Some(tags) = &result.tags { - final_log = final_log.tags(tags.clone()); - } - if !result.errors.is_empty() { - final_log = final_log.error( - serde_json::to_value(&result.errors) - .map_err(|error| crate::BraintrustError::Background(error.to_string()))?, - ); - } - let final_write = case_span.log(final_log.build().unwrap()); - let end = case_span.end(); - let initial_result = initial_write.await; - let final_result = final_write.await; - let end_result = end.await; - initial_result?; - final_result?; - end_result?; - - if !quiet { - let score_str = result - .scores - .iter() - .map(|s| format!("{}={:.3}", s.name, s.score)) - .collect::>() - .join(", "); - println!("Case {}: {}", case_num, score_str); - for error in &result.errors { - eprintln!("Case {}: {}", case_num, error); - } - } - Ok(result) + let name = opts.name.clone(); + let metadata = opts.metadata.clone(); + let tags = opts.tags.clone(); + let mut reporter = super::reporter::TerminalReporter { quiet: opts.quiet }; + let plan = super::executor::EvalPlan::new(opts)?; + let recorder = + super::recorder::BraintrustRecorder::new(self.client.clone(), name, metadata, tags) + .await?; + super::executor::EvalExecutor::run(plan, Some(&recorder), &mut reporter).await } } diff --git a/src/eval/types.rs b/src/eval/types.rs index 9078148..8b34da0 100644 --- a/src/eval/types.rs +++ b/src/eval/types.rs @@ -155,9 +155,9 @@ pub struct EvalSummary { pub score_stats: Vec, /// Total number of cases evaluated pub total_cases: usize, - /// Number of successful evaluations + /// Cases without task/scorer errors, independently of recording status. pub successful_cases: usize, - /// Number of failed evaluations + /// Cases with task/scorer errors, independently of recording status. pub failed_cases: usize, /// Full dataset provenance captured before source consumption. #[serde(skip_serializing_if = "Option::is_none")] diff --git a/tests/eval_test.rs b/tests/eval_test.rs index 0c6fdc5..68e9a42 100644 --- a/tests/eval_test.rs +++ b/tests/eval_test.rs @@ -554,8 +554,21 @@ async fn recording_failure_stops_new_cases_but_drains_already_scheduled_cases() ))); let partial = error.partial.unwrap(); assert_eq!(partial.total_cases, 2); - assert_eq!(partial.failed_cases, 2); - assert_eq!(partial.successful_cases, 0); + assert_eq!(partial.failed_cases, 0); + assert_eq!(partial.successful_cases, 2); + assert_eq!(partial.results.len(), 2); + assert_eq!( + partial + .results + .iter() + .map(|result| result.output) + .collect::>(), + [Some(0), Some(1)] + ); + assert!(partial + .results + .iter() + .all(|result| result.errors.is_empty() && result.span_id.is_some())); } #[tokio::test] @@ -614,3 +627,76 @@ async fn reusable_definition_keeps_recorded_runs_and_sources_independent() { assert_eq!(later.results[0].output, Some(10)); assert_eq!(calls.load(Ordering::SeqCst), 3); } + +#[tokio::test] +async fn serialization_failures_keep_outputs_and_finalize_recording() { + #[derive(Debug)] + struct Value(usize); + impl serde::Serialize for Value { + fn serialize(&self, _: S) -> std::result::Result { + Err(serde::ser::Error::custom( + "application serialization failed", + )) + } + } + let (client, server) = recording_client().await; + let options = EvalOpts::builder() + .name("serialization-failure".into()) + .source( + vec![Case { + input: Value(7), + expected: Some(Value(14)), + metadata: None, + tags: None, + id: None, + xact_id: None, + created: None, + }] + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &Value| Ok(Value(input.0 * 2)))), + vec![], + ) + .unwrap(), + ) + .quiet(true) + .build(); + let error = Evaluator::new(client).run(options).await.unwrap_err(); + let fields = error + .failures + .iter() + .map(|failure| match failure { + EvalRunFailure::Recording { + case_index: Some(0), + source, + } => source.to_string(), + other => panic!("unexpected failure: {other}"), + }) + .collect::>(); + assert_eq!(fields.len(), 3); + for (field, message) in ["input", "expected", "output"].into_iter().zip(fields) { + assert!(message.contains(field)); + assert!(message.contains("application serialization failed")); + } + let summary = error.partial.unwrap(); + assert_eq!(summary.successful_cases, 1); + assert_eq!(summary.failed_cases, 0); + assert_eq!(summary.results[0].output.as_ref().unwrap().0, 14); + assert!(summary.results[0].span_id.is_some()); + let requests = server.received_requests().await.unwrap(); + let ended: std::collections::HashSet<_> = requests + .iter() + .filter(|request| request.url.path() == "/logs3") + .flat_map(|request| { + serde_json::from_slice::(&request.body).unwrap()["rows"] + .as_array() + .unwrap() + .clone() + }) + .filter(|row| row["metrics"]["end"].is_number()) + .map(|row| row["id"].as_str().unwrap().to_owned()) + .collect(); + assert_eq!(ended.len(), 2, "both case and run recording must finalize"); +}