diff --git a/src/eval/bt_runner.rs b/src/eval/bt_runner.rs index 8662a52..b775461 100644 --- a/src/eval/bt_runner.rs +++ b/src/eval/bt_runner.rs @@ -16,8 +16,7 @@ use serde_json::{json, Value}; use crate::{BraintrustClient, BraintrustError, Result}; use super::{ - runner::{EvalOpts, Evaluator}, - types::EvalSummary, + runner::EvalOpts, types::EvalSummary, EvalReporter, EvalRequest, JsonlReporter, ReportError, }; /// A filter for selecting which evaluators to run, read from `BT_EVAL_FILTER_PARSED`. @@ -115,12 +114,27 @@ struct SseWriter { } impl SseWriter { - fn send(&self, event: &str, data: &Value) { - let data_str = serde_json::to_string(data).unwrap_or_default(); - if let Ok(mut w) = self.writer.lock() { - let _ = write!(w, "event: {}\ndata: {}\n\n", event, data_str); - let _ = w.flush(); - } + fn send(&self, event: &str, data: &Value) -> std::result::Result<(), ReportError> { + let data_str = serde_json::to_string(data)?; + let mut writer = self + .writer + .lock() + .map_err(|_| std::io::Error::other("SSE writer lock poisoned"))?; + write!(writer, "event: {}\ndata: {}\n\n", event, data_str)?; + writer.flush()?; + Ok(()) + } +} + +struct SseReporter { + sse: Arc, +} +impl EvalReporter for SseReporter { + fn started(&mut self, name: &str) -> std::result::Result<(), ReportError> { + self.sse.send("start", &json!({"name": name})) + } + fn finished(&mut self, summary: &EvalSummary) -> std::result::Result<(), ReportError> { + self.sse.send("summary", &summary_to_value(summary)?) } } @@ -224,14 +238,17 @@ pub struct BtEvalRunner { impl BtEvalRunner { /// Create a new runner, reading configuration from environment variables. /// - /// In non-list mode, this creates a `BraintrustClient` from environment variables - /// (e.g. `BRAINTRUST_API_KEY`). In list mode, no client is created. + /// Recording mode creates a client from environment variables (e.g. + /// `BRAINTRUST_API_KEY`). List and local modes create no client. pub async fn from_env() -> Result { let config = RunnerConfig::from_env(); let sse = create_sse_writer(); - // Only create a client when we'll actually run evals - let client = if !config.list { + Self::from_config(config, sse).await + } + + async fn from_config(config: RunnerConfig, sse: Option>) -> Result { + let client = if !config.list && !config.no_send_logs { Some(Arc::new(BraintrustClient::builder().build().await?)) } else { None @@ -311,30 +328,27 @@ impl BtEvalRunner { opts.project_name = Some(project.to_string()); } - let client = self - .client - .clone() - .ok_or_else(|| BraintrustError::Background("BtEvalRunner has no client".to_string()))?; - - let evaluator = Evaluator::::from_arc(client); - - if let Some(sse) = &self.sse { - sse.send("start", &json!({ "name": name })); - } + let request = if self.config.no_send_logs { + EvalRequest::local(opts) + } else { + let client = self.client.as_ref().ok_or_else(|| { + BraintrustError::Background("BtEvalRunner has no client".to_string()) + })?; + EvalRequest::recorded(opts, client.as_ref().clone()) + }; + // Protocol output replaces terminal progress even when quiet is false. + let request = if let Some(sse) = &self.sse { + request.with_reporter(SseReporter { sse: sse.clone() }) + } else if self.config.jsonl { + request.with_reporter(JsonlReporter::new(std::io::stdout())) + } else { + request + }; - match evaluator.run(opts).await { + match super::run(request).await { Ok(summary) => { let failed = summary.failed_cases; - if let Some(sse) = &self.sse { - let summary_value = serde_json::to_value(&summary).unwrap_or(Value::Null); - sse.send("summary", &summary_value); - } else if self.config.jsonl { - if let Ok(json) = serde_json::to_string(&summary) { - println!("{}", json); - } - } - if failed > 0 { self.all_passed = false; if self.config.terminate_on_failure { @@ -347,8 +361,17 @@ impl BtEvalRunner { } Err(e) => { self.all_passed = false; - if let Some(sse) = &self.sse { - sse.send("error", &json!({ "message": e.to_string() })); + let output_failed = e + .failures + .iter() + .any(|failure| matches!(failure, super::EvalRunFailure::Reporting { .. })); + if let Some(sse) = self.sse.as_ref().filter(|_| !output_failed) { + if let Err(output) = sse.send("error", &json!({ "message": e.to_string() })) { + eprintln!( + "Evaluator '{}' failed: {}; error output also failed: {}", + name, e, output + ); + } } else { eprintln!("Evaluator '{}' failed: {}", name, e); } @@ -394,13 +417,13 @@ impl BtEvalRunner { /// Serialize an [`EvalSummary`] in a way compatible with the SSE summary event. /// -/// This is used internally by [`BtEvalRunner`] and is public only for advanced use cases. -pub fn summary_to_value(summary: &EvalSummary) -> Value +/// Serialization failures are returned with their cause; no null placeholder is emitted. +pub fn summary_to_value(summary: &EvalSummary) -> serde_json::Result where I: Serialize, O: Serialize, { - serde_json::to_value(summary).unwrap_or(Value::Null) + serde_json::to_value(summary) } #[cfg(test)] @@ -469,4 +492,121 @@ mod tests { assert!(filter.matches("foo-eval")); assert!(!filter.matches("bar-eval")); } + #[derive(Clone, Default)] + struct Buffer(Arc>>); + impl Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + fn local_config(terminate_on_failure: bool) -> RunnerConfig { + RunnerConfig { + jsonl: true, + list: false, + terminate_on_failure, + filters: vec![], + no_send_logs: true, + } + } + fn local_options(fail: bool) -> EvalOpts { + super::super::EvalOpts::builder() + .name("cli".into()) + .quiet(false) + .source( + vec![super::super::Case { + input: 1, + ..Default::default() + }] + .into(), + ) + .definition( + super::super::EvalDefinition::new( + Box::new(super::super::SimpleFnTask::new(move |input: &usize| { + if fail { + anyhow::bail!("case failed") + } else { + Ok(*input) + } + })), + vec![], + ) + .unwrap(), + ) + .build() + } + #[tokio::test] + async fn sse_protocol_is_the_only_reporter_and_case_failures_update_cli_status() { + let output = Buffer::default(); + let sse = Arc::new(SseWriter { + writer: Mutex::new(Box::new(output.clone())), + }); + let mut runner = BtEvalRunner::from_config(local_config(false), Some(sse)) + .await + .unwrap(); + assert!(runner.client.is_none()); + runner.eval("project", local_options(false)).await.unwrap(); + runner.eval("project", local_options(true)).await.unwrap(); + assert!(!runner.all_passed()); + assert!(!runner.finish().await.unwrap()); + let bytes = output.0.lock().unwrap(); + let text = std::str::from_utf8(&bytes).unwrap(); + let events: Vec<_> = text.split("\n\n").filter(|s| !s.is_empty()).collect(); + assert_eq!(events.len(), 4); + assert!(events[0].starts_with("event: start\ndata: ")); + assert!(events[1].starts_with("event: summary\ndata: ")); + assert!(events[2].starts_with("event: start\ndata: ")); + let summary: Value = + serde_json::from_str(events[3].strip_prefix("event: summary\ndata: ").unwrap()) + .unwrap(); + assert_eq!(summary["failed_cases"], 1); + } + #[tokio::test] + async fn terminate_on_failure_returns_an_error_and_retains_failed_session_status() { + let sse = Arc::new(SseWriter { + writer: Mutex::new(Box::new(Buffer::default())), + }); + let mut runner = BtEvalRunner::from_config(local_config(true), Some(sse)) + .await + .unwrap(); + assert!(runner.eval("project", local_options(true)).await.is_err()); + assert!(!runner.all_passed()); + assert!(!runner.finish().await.unwrap()); + } + #[tokio::test] + async fn sse_write_failure_is_a_report_failure_without_losing_computation() { + struct Broken; + impl Write for Broken { + fn write(&mut self, _: &[u8]) -> std::io::Result { + Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "closed socket", + )) + } + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + let sse = Arc::new(SseWriter { + writer: Mutex::new(Box::new(Broken)), + }); + let result = super::super::run( + EvalRequest::local(local_options(false)) + .with_reporter(SseReporter { sse: sse.clone() }), + ) + .await; + let error = result.unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [super::super::EvalRunFailure::Reporting { + stage: super::super::ReportStage::Started, + source: ReportError::Io(_), + .. + }] + )); + assert_eq!(error.partial.unwrap().successful_cases, 1); + } } diff --git a/src/eval/error.rs b/src/eval/error.rs index 37b11dc..694d9df 100644 --- a/src/eval/error.rs +++ b/src/eval/error.rs @@ -71,6 +71,13 @@ pub enum EvalRunFailure { #[source] source: crate::BraintrustError, }, + #[error("reporting failed at {stage:?} for case {case_index:?}: {source}")] + Reporting { + stage: super::ReportStage, + case_index: Option, + #[source] + source: super::ReportError, + }, #[error(transparent)] Execution(#[from] crate::ExecutionError), #[error(transparent)] @@ -79,7 +86,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. +/// Recording and reporting failures preserve computed results and their success counts. pub struct EvalRunError { pub failures: Vec, pub partial: Option>>, diff --git a/src/eval/executor.rs b/src/eval/executor.rs index af24c3f..40686e7 100644 --- a/src/eval/executor.rs +++ b/src/eval/executor.rs @@ -9,6 +9,7 @@ use super::{ case_runner::CaseRunner, Case, EvalDefinition, EvalOpts, EvalResult, EvalRunError, EvalRunFailure, EvalSummary, }; +use super::{reporter::ReporterSession, EvalReporter}; use crate::{BraintrustError, Result, SpanHandle}; /// An invocation owns its source and options; only the definition is shared. @@ -71,12 +72,6 @@ pub(crate) trait CaseRecording: Send { 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, @@ -87,9 +82,10 @@ impl EvalExecutor { pub async fn run( mut plan: EvalPlan, mut recorder: Option<&mut dyn EvalRecorder>, - reporter: &mut impl EvalReporter, + reporter: &mut dyn EvalReporter, binding: Option<&crate::internal::execution::ExecutionBinding>, ) -> std::result::Result, EvalRunError> { + let mut reporter = ReporterSession::new(reporter); reporter.started(&plan.name); let prepared = plan.source.prepare().await; let dataset = plan.source.provenance(); @@ -126,7 +122,13 @@ impl EvalExecutor { 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); + Self::collect( + index, + outcome, + &mut results, + &mut failures, + &mut reporter, + ); if !failures.is_empty() { break Ok(None); } @@ -158,7 +160,7 @@ impl EvalExecutor { }); if pending.len() >= plan.concurrency { if let Some((index, outcome)) = pending.next().await { - Self::collect(index, outcome, &mut results, &mut failures, reporter); + Self::collect(index, outcome, &mut results, &mut failures, &mut reporter); } if !failures.is_empty() { break; @@ -167,7 +169,7 @@ impl EvalExecutor { } // 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); + Self::collect(index, outcome, &mut results, &mut failures, &mut reporter); } if let Some(recorder) = recorder { failures.extend(recorder.finish().await.into_iter().map(|source| { @@ -194,7 +196,9 @@ impl EvalExecutor { dataset, recording: recorder.and_then(|recorder| recorder.target()), }; - reporter.finished(&summary); + if let Some(failure) = reporter.finished(&summary) { + failures.push(failure); + } if failures.is_empty() { Ok(summary) } else { @@ -237,7 +241,7 @@ impl EvalExecutor { outcome: CaseOutcome, results: &mut Vec<(usize, EvalResult)>, failures: &mut Vec, - reporter: &mut impl EvalReporter, + reporter: &mut ReporterSession<'_, I, O>, ) { failures.extend( outcome @@ -284,12 +288,20 @@ mod tests { finished: bool, } impl EvalReporter for Observed { - fn started(&mut self, _: &str) {} - fn case_finished(&mut self, index: usize, _: &EvalResult) { + fn case_finished( + &mut self, + index: usize, + _: &EvalResult, + ) -> std::result::Result<(), super::super::ReportError> { self.cases.push(index); + Ok(()) } - fn finished(&mut self, _: &EvalSummary) { + fn finished( + &mut self, + _: &EvalSummary, + ) -> std::result::Result<(), super::super::ReportError> { self.finished = true; + Ok(()) } } struct WaitingSource { diff --git a/src/eval/mod.rs b/src/eval/mod.rs index 8a556b3..2306503 100644 --- a/src/eval/mod.rs +++ b/src/eval/mod.rs @@ -69,6 +69,9 @@ pub use error::{CaseError, EvalConfigError, EvalRunError, EvalRunFailure}; // Re-export public API pub use bt_runner::{summary_to_value, BtEvalRunner}; +pub use reporter::{ + EvalReporter, JsonlReporter, ReportError, ReportStage, SilentReporter, TerminalReporter, +}; pub use run::{run, EvalRequest}; pub use runner::{EvalOpts, Evaluator}; pub use scorer::{scorer, ExactMatch, LevenshteinScorer, NumericDiffScorer, Scorer, SyncFnScorer}; diff --git a/src/eval/reporter.rs b/src/eval/reporter.rs index 149fce5..aadeb4f 100644 --- a/src/eval/reporter.rs +++ b/src/eval/reporter.rs @@ -1,44 +1,197 @@ -//! Console presentation, separate from execution and recording. -use super::{executor::EvalReporter, EvalResult, EvalSummary}; +//! Run-owned output adapters. Reporting never owns computation or recording. +use std::io::{self, Write}; -pub(crate) struct TerminalReporter { - pub quiet: bool, +use serde::Serialize; + +use super::{EvalResult, EvalRunFailure, EvalSummary}; + +/// A failure to present evaluation progress or results. +#[derive(Debug, thiserror::Error)] +pub enum ReportError { + #[error("output failed: {0}")] + Io(#[from] io::Error), + #[error("result serialization failed: {0}")] + Serialization(#[from] serde_json::Error), + #[error(transparent)] + Other(#[from] anyhow::Error), } -impl EvalReporter for TerminalReporter { - fn started(&mut self, name: &str) { - if !self.quiet { - println!("Running evaluation: {name}"); - } + +/// Where a reporter failed. Case indices in run failures are zero-based. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ReportStage { + Started, + CaseFinished, + Finished, +} + +/// Per-run presentation, without serialization bounds on local computation. +/// +/// Callbacks run sequentially on the run's execution binding and must return +/// promptly. Case notifications follow completion order and carry source indices; +/// the final summary is in source order. `finished` also receives partial summaries +/// after source or recording failures. +/// +/// The first returned error disables this reporter for the rest of the run. +/// Computation and recording continue; the completion reports a `Reporting` +/// failure with the computed summary in `EvalRunError::partial`. Reporter panics +/// follow the run's ordinary worker-loss behavior. +pub trait EvalReporter: Send { + fn started(&mut self, _name: &str) -> Result<(), ReportError> { + Ok(()) + } + fn case_finished( + &mut self, + _index: usize, + _result: &EvalResult, + ) -> Result<(), ReportError> { + Ok(()) + } + fn finished(&mut self, _summary: &EvalSummary) -> Result<(), ReportError> { + Ok(()) + } +} + +/// Suppress all progress and result output. +pub struct SilentReporter; +impl EvalReporter for SilentReporter {} + +/// Human-readable progress on stdout and case errors on stderr. +pub struct TerminalReporter { + output: Box, + errors: Box, +} +impl Default for TerminalReporter { + fn default() -> Self { + Self::with_writers(io::stdout(), io::stderr()) } - fn case_finished(&mut self, index: usize, result: &EvalResult) { - if self.quiet { - return; +} +impl TerminalReporter { + pub fn with_writers( + output: impl Write + Send + 'static, + errors: impl Write + Send + 'static, + ) -> Self { + Self { + output: Box::new(output), + errors: Box::new(errors), } + } +} +impl EvalReporter for TerminalReporter { + fn started(&mut self, name: &str) -> Result<(), ReportError> { + writeln!(self.output, "Running evaluation: {name}")?; + Ok(()) + } + fn case_finished( + &mut self, + index: usize, + result: &EvalResult, + ) -> Result<(), ReportError> { let scores = result .scores .iter() .map(|score| format!("{}={:.3}", score.name, score.score)) .collect::>() .join(", "); - println!("Case {}: {}", index + 1, scores); + writeln!(self.output, "Case {}: {}", index + 1, scores)?; for error in &result.errors { - eprintln!("Case {}: {}", index + 1, error); + writeln!(self.errors, "Case {}: {}", index + 1, error)?; } + Ok(()) } - 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:"); + fn finished(&mut self, summary: &EvalSummary) -> Result<(), ReportError> { + writeln!(self.output, "\nEvaluation results:")?; + writeln!(self.output, "Total cases: {}", summary.total_cases)?; + writeln!(self.output, "Successful: {}", summary.successful_cases)?; + writeln!(self.output, "Failed: {}", summary.failed_cases)?; + writeln!(self.output, "\nScore summary:")?; for stat in &summary.score_stats { - println!( + writeln!( + self.output, " {}: mean={:.3}, min={:.3}, max={:.3}, count={}", stat.name, stat.mean, stat.min, stat.max, stat.count - ); + )?; + } + self.output.flush()?; + self.errors.flush()?; + Ok(()) + } +} + +/// Emit one serialized summary followed by a newline, with no progress text. +/// Serialization is required only when this adapter is selected. Serialization +/// errors are returned before writing a line; I/O failures may leave a partial line. +/// +/// ```no_run +/// use braintrust_sdk_rust::eval::{EvalOpts, EvalRequest, JsonlReporter}; +/// # fn example(options: EvalOpts) { +/// let completion = braintrust_sdk_rust::eval::run( +/// EvalRequest::local(options).with_reporter(JsonlReporter::new(std::io::stdout())), +/// ); +/// # } +/// ``` +pub struct JsonlReporter { + writer: W, +} +impl JsonlReporter { + pub fn new(writer: W) -> Self { + Self { writer } + } +} +impl EvalReporter for JsonlReporter { + fn finished(&mut self, summary: &EvalSummary) -> Result<(), ReportError> { + let line = serde_json::to_vec(summary)?; + self.writer.write_all(&line)?; + self.writer.write_all(b"\n")?; + self.writer.flush()?; + Ok(()) + } +} + +/// Keep reporting failures out of the scheduler's stop-reading condition. +/// Disable after the first error so broken output cannot accumulate errors per case. +pub(crate) struct ReporterSession<'a, I, O> { + reporter: Option<&'a mut dyn EvalReporter>, + failure: Option, +} +impl<'a, I, O> ReporterSession<'a, I, O> { + pub fn new(reporter: &'a mut dyn EvalReporter) -> Self { + Self { + reporter: Some(reporter), + failure: None, + } + } + fn observe( + &mut self, + outcome: Result<(), ReportError>, + stage: ReportStage, + case_index: Option, + ) { + if let Err(source) = outcome { + self.reporter = None; + self.failure = Some(EvalRunFailure::Reporting { + stage, + case_index, + source, + }); + } + } + pub fn started(&mut self, name: &str) { + if let Some(reporter) = self.reporter.as_mut() { + let outcome = reporter.started(name); + self.observe(outcome, ReportStage::Started, None); + } + } + pub fn case_finished(&mut self, index: usize, result: &EvalResult) { + if let Some(reporter) = self.reporter.as_mut() { + let outcome = reporter.case_finished(index, result); + self.observe(outcome, ReportStage::CaseFinished, Some(index)); + } + } + pub fn finished(&mut self, summary: &EvalSummary) -> Option { + if let Some(reporter) = self.reporter.as_mut() { + let outcome = reporter.finished(summary); + self.observe(outcome, ReportStage::Finished, None); } + self.failure.take() } } diff --git a/src/eval/run.rs b/src/eval/run.rs index 7d1eb4e..04ab6b6 100644 --- a/src/eval/run.rs +++ b/src/eval/run.rs @@ -2,14 +2,16 @@ use super::{ executor::{EvalExecutor, EvalPlan}, recorder::BraintrustRecorder, - reporter::TerminalReporter, - EvalConfigError, EvalOpts, EvalRunError, EvalRunFailure, EvalSummary, + reporter::{SilentReporter, TerminalReporter}, + EvalConfigError, EvalOpts, EvalReporter, EvalRunError, EvalRunFailure, EvalSummary, }; use crate::{internal::execution::ExecutionBinding, BraintrustClient, Completion}; use serde::Serialize; type RunCompletion = Completion, EvalRunError>; -type SubmitRecorded = fn(BraintrustClient, EvalOpts) -> RunCompletion; +type RunReporter = Box>; +type SubmitRecorded = + fn(BraintrustClient, EvalOpts, RunReporter) -> RunCompletion; /// An owned invocation of a reusable definition with its own source and options. /// @@ -17,6 +19,7 @@ type SubmitRecorded = fn(BraintrustClient, EvalOpts) -> RunCompletio /// serialization and inherit their client's execution binding and lifecycle. pub struct EvalRequest { options: EvalOpts, + reporter: Option>, target: Target, } enum Target { @@ -27,10 +30,17 @@ enum Target { }, } impl EvalRequest { + /// Select run-owned output. This overrides `EvalOpts::quiet`, which only + /// controls the default terminal reporter. Reporter failures preserve results. + pub fn with_reporter(mut self, reporter: impl EvalReporter + 'static) -> Self { + self.reporter = Some(Box::new(reporter)); + self + } /// Run without login, HTTP or recording, on `options.runtime` or an owned runtime. pub fn local(options: EvalOpts) -> Self { Self { options, + reporter: None, target: Target::Local, } } @@ -40,6 +50,7 @@ impl pub fn recorded(options: EvalOpts, client: BraintrustClient) -> Self { Self { options, + reporter: None, target: Target::Recorded { client, submit: recorded::, @@ -73,6 +84,13 @@ pub fn run( if let Err(error) = request.options.validate() { return Completion::rejected(error.into()); } + let mut reporter = request.reporter.unwrap_or_else(|| { + if request.options.quiet { + Box::new(SilentReporter) + } else { + Box::new(TerminalReporter::default()) + } + }); match request.target { Target::Local => { let mut options = request.options; @@ -84,13 +102,10 @@ pub fn run( }; let callbacks = binding.clone(); binding.response(async move { - let mut reporter = TerminalReporter { - quiet: options.quiet, - }; EvalExecutor::run( EvalPlan::new(options)?, None, - &mut reporter, + reporter.as_mut(), Some(&callbacks), ) .await @@ -100,7 +115,7 @@ pub fn run( if request.options.runtime.is_some() { return Completion::rejected(EvalConfigError::RecordedRuntime.into()); } - submit(client, request.options) + submit(client, request.options, reporter) } } } @@ -108,15 +123,13 @@ pub fn run( fn recorded( client: BraintrustClient, options: EvalOpts, + mut reporter: RunReporter, ) -> RunCompletion { let binding = client.execution_binding(); let callbacks = binding.clone(); binding.response_try(|| { let operation = client.span_operation()?; Ok(async move { - let mut reporter = TerminalReporter { - quiet: options.quiet, - }; let mut recorder = BraintrustRecorder::new( client, operation.submitter.clone(), @@ -130,7 +143,7 @@ fn recorded { /// Maximum concurrent evaluations (default: 5) #[builder(default = 5)] pub max_concurrency: usize, - /// Suppress progress output (default: false) + /// Suppress default terminal output (default: false). An explicit request reporter takes precedence. #[builder(default = false)] pub quiet: bool, /// Explicit runtime for local runs. Recorded runs inherit their client binding. diff --git a/tests/eval_reporting.rs b/tests/eval_reporting.rs new file mode 100644 index 0000000..bd47379 --- /dev/null +++ b/tests/eval_reporting.rs @@ -0,0 +1,516 @@ +use braintrust_sdk_rust::{eval::*, BraintrustClient}; +use serde::{Serialize, Serializer}; +use serde_json::json; +use std::{ + cell::Cell, + io::{self, Write}, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, Mutex, + }, + time::Duration, +}; + +#[derive(Clone, Default)] +struct Buffer(Arc>>); +impl Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } +} +fn options(calls: Arc) -> EvalOpts { + EvalOpts::builder() + .name("reporting".into()) + .quiet(true) + .max_concurrency(1) + .source( + (0..4) + .map(|input| Case { + input, + ..Default::default() + }) + .collect::>() + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(move |input: &usize| { + calls.fetch_add(1, Ordering::SeqCst); + Ok(input * 2) + })), + vec![], + ) + .unwrap(), + ) + .build() +} +struct FailedReporter { + stage: ReportStage, + attempts: Arc, +} +impl FailedReporter { + fn report(&self, stage: ReportStage) -> Result<(), ReportError> { + self.attempts.fetch_add(1, Ordering::SeqCst); + if self.stage == stage { + Err(io::Error::new(io::ErrorKind::BrokenPipe, "closed output").into()) + } else { + Ok(()) + } + } +} +impl EvalReporter for FailedReporter { + fn started(&mut self, _: &str) -> Result<(), ReportError> { + self.report(ReportStage::Started) + } + fn case_finished(&mut self, _: usize, _: &EvalResult) -> Result<(), ReportError> { + self.report(ReportStage::CaseFinished) + } + fn finished(&mut self, _: &EvalSummary) -> Result<(), ReportError> { + self.report(ReportStage::Finished) + } +} +#[test] +fn reporting_errors_disable_output_without_stopping_cases_or_losing_results() { + for stage in [ + ReportStage::Started, + ReportStage::CaseFinished, + ReportStage::Finished, + ] { + let calls = Arc::new(AtomicUsize::new(0)); + let attempts = Arc::new(AtomicUsize::new(0)); + let completion = run(EvalRequest::local(options(calls.clone())).with_reporter( + FailedReporter { + stage, + attempts: attempts.clone(), + }, + )); + assert!(completion.is_ok()); + let Err(error) = futures::executor::block_on(async { completion.await }) else { + panic!("reporting must fail"); + }; + let [EvalRunFailure::Reporting { + stage: actual, + case_index, + source: ReportError::Io(source), + }] = error.failures.as_slice() + else { + panic!("unexpected failures") + }; + assert_eq!(*actual, stage); + assert_eq!( + *case_index, + (stage == ReportStage::CaseFinished).then_some(0) + ); + assert_eq!(source.kind(), io::ErrorKind::BrokenPipe); + assert_eq!( + attempts.load(Ordering::SeqCst), + match stage { + ReportStage::Started => 1, + ReportStage::CaseFinished => 2, + ReportStage::Finished => 6, + } + ); + assert_eq!(calls.load(Ordering::SeqCst), 4); + let summary = error.partial.unwrap(); + assert_eq!(summary.successful_cases, 4); + assert_eq!( + summary + .results + .iter() + .map(|r| r.output.unwrap()) + .collect::>(), + [0, 2, 4, 6] + ); + } +} +#[test] +fn custom_local_reporter_needs_no_serialization_clone_debug_or_sync() { + struct Input(usize); + struct Output(usize); + struct Reporter { + count: Cell, + observed: Arc, + } + impl EvalReporter for Reporter { + fn case_finished( + &mut self, + index: usize, + result: &EvalResult, + ) -> Result<(), ReportError> { + assert_eq!(index, 0); + assert_eq!(result.output.as_ref().unwrap().0, 9); + self.count.set(self.count.get() + 1); + Ok(()) + } + fn finished(&mut self, _: &EvalSummary) -> Result<(), ReportError> { + self.observed.store(self.count.get(), Ordering::SeqCst); + Ok(()) + } + } + let observed = Arc::new(AtomicUsize::new(0)); + let opts = EvalOpts::builder() + .name("opaque".into()) + .source( + vec![Case { + input: Input(3), + expected: None, + metadata: None, + tags: None, + id: None, + xact_id: None, + created: None, + }] + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &Input| Ok(Output(input.0 * 3)))), + vec![], + ) + .unwrap(), + ) + .build(); + let completion = run(EvalRequest::local(opts).with_reporter(Reporter { + count: Cell::new(0), + observed: observed.clone(), + })); + futures::executor::block_on(async { completion.await }).unwrap(); + assert_eq!(observed.load(Ordering::SeqCst), 1); +} +#[test] +fn jsonl_outputs_one_complete_summary_and_terminal_uses_separate_writers() { + let output = Buffer::default(); + let completion = run(EvalRequest::local(options(Arc::new(AtomicUsize::new(0)))) + .with_reporter(JsonlReporter::new(output.clone()))); + futures::executor::block_on(async { completion.await }).unwrap(); + let bytes = output.0.lock().unwrap(); + assert_eq!(bytes.iter().filter(|&&b| b == b'\n').count(), 1); + let summary: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(summary["total_cases"], 4); + drop(bytes); + let terminal = Buffer::default(); + let errors = Buffer::default(); + let mut opts = options(Arc::new(AtomicUsize::new(0))); + opts.definition = EvalDefinition::new( + Box::new(SimpleFnTask::new(|_: &usize| -> anyhow::Result { + anyhow::bail!("task-error") + })), + vec![], + ) + .unwrap(); + let completion = run( + EvalRequest::local(opts).with_reporter(TerminalReporter::with_writers( + terminal.clone(), + errors.clone(), + )), + ); + assert_eq!( + futures::executor::block_on(async { completion.await }) + .unwrap() + .failed_cases, + 4 + ); + assert!(String::from_utf8(terminal.0.lock().unwrap().clone()) + .unwrap() + .contains("Failed: 4")); + assert!(String::from_utf8(errors.0.lock().unwrap().clone()) + .unwrap() + .contains("task-error")); +} +struct Unserializable; +impl Serialize for Unserializable { + fn serialize(&self, _: S) -> Result { + Err(serde::ser::Error::custom("cannot serialize")) + } +} +#[test] +fn serialization_errors_preserve_computed_results_without_writing_fake_json() { + let output = Buffer::default(); + let opts = EvalOpts::builder() + .name("serialize".into()) + .source( + vec![Case { + input: Unserializable, + expected: None, + metadata: None, + tags: None, + id: None, + xact_id: None, + created: None, + }] + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|_: &Unserializable| Ok(7usize))), + vec![], + ) + .unwrap(), + ) + .build(); + let completion = + run(EvalRequest::local(opts).with_reporter(JsonlReporter::new(output.clone()))); + let Err(error) = futures::executor::block_on(async { completion.await }) else { + panic!("reporting must fail"); + }; + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Reporting { + stage: ReportStage::Finished, + source: ReportError::Serialization(_), + .. + }] + )); + let partial = error.partial.unwrap(); + assert!(summary_to_value(&partial).is_err()); + assert_eq!(partial.results[0].output, Some(7)); + assert!(output.0.lock().unwrap().is_empty()); +} +#[tokio::test] +async fn reporter_failure_preserves_recorded_results_and_does_not_poison_shutdown() { + use wiremock::{ + matchers::{method, path}, + Mock, MockServer, ResponseTemplate, + }; + let server = MockServer::start().await; + for (endpoint, body) in [ + ( + "/api/apikey/login", + json!({"org_info":[{"id":"org","name":"org"}]}), + ), + ( + "/api/experiment/register", + json!({"project":{"id":"project","name":"project"},"experiment":{"id":"experiment","name":"experiment"}}), + ), + ("/logs3", json!({})), + ] { + Mock::given(method("POST")) + .and(path(endpoint)) + .respond_with(ResponseTemplate::new(200).set_body_json(body)) + .mount(&server) + .await; + } + Mock::given(method("GET")) + .and(path("/version")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({}))) + .mount(&server) + .await; + let client = BraintrustClient::builder() + .runtime(braintrust_sdk_rust::current_runtime()) + .api_key("key") + .app_url(server.uri()) + .api_url(server.uri()) + .default_project("project") + .blocking_login(true) + .build() + .await + .unwrap(); + let calls = Arc::new(AtomicUsize::new(0)); + let completion = run( + EvalRequest::recorded(options(calls.clone()), client.clone()).with_reporter( + FailedReporter { + stage: ReportStage::Started, + attempts: Arc::new(AtomicUsize::new(0)), + }, + ), + ); + let shutdown = client.shutdown(Duration::from_secs(5)); + let (result, shutdown) = tokio::join!(completion, shutdown); + shutdown.unwrap(); + let error = result.unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Reporting { .. }] + )); + let partial = error.partial.unwrap(); + assert_eq!(partial.recording.unwrap().experiment_id, "experiment"); + assert_eq!(partial.successful_cases, 4); + let requests = server.received_requests().await.unwrap(); + let rows: Vec = requests + .iter() + .filter(|r| r.url.path() == "/logs3") + .flat_map(|r| { + serde_json::from_slice::(&r.body).unwrap()["rows"] + .as_array() + .unwrap() + .clone() + }) + .collect(); + for case in partial.results { + assert!(rows + .iter() + .any(|r| r["span_id"] == *case.span_id.as_ref().unwrap() + && r["metrics"]["end"].is_number())); + } + assert_eq!(calls.load(Ordering::SeqCst), 4); +} + +// A subprocess exercises actual environment parsing and stdout without mutating +// the test runner's process-wide environment or racing other tests. +#[test] +fn cli_output_child() { + let Ok(mode) = std::env::var("BT_REPORTING_CHILD") else { + return; + }; + futures::executor::block_on(async { + let mut runner = BtEvalRunner::from_env().await.unwrap(); + assert!(runner.no_send_logs()); + let calls = Arc::new(AtomicUsize::new(0)); + for name in ["excluded", "selected"] { + let mut opts = options(calls.clone()); + opts.name = name.into(); + opts.quiet = false; + // No recording span is provided in local CLI mode. + opts.definition = EvalDefinition::new( + Box::new(task({ + let calls = calls.clone(); + move |input: &usize, hooks| { + assert!(hooks.span.is_none()); + calls.fetch_add(1, Ordering::SeqCst); + Box::pin(async move { Ok(*input) }) + } + })), + vec![], + ) + .unwrap(); + runner.eval("project", opts).await.unwrap(); + } + assert!(runner.finish().await.unwrap()); + assert_eq!( + calls.load(Ordering::SeqCst), + if mode == "list" { 0 } else { 4 } + ); + }); +} +#[test] +fn cli_local_jsonl_and_list_modes_need_no_credentials_and_keep_stdout_clean() { + for mode in ["jsonl", "no-send", "list"] { + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "cli_output_child", "--nocapture"]) + .env("BT_REPORTING_CHILD", mode) + .env("BT_EVAL_LOCAL", if mode == "no-send" { "0" } else { "1" }) + .env( + "BT_EVAL_NO_SEND_LOGS", + if mode == "no-send" { "1" } else { "0" }, + ) + .env("BT_EVAL_LIST", if mode == "list" { "1" } else { "0" }) + .env("BT_EVAL_JSONL", "1") + .env("BT_EVAL_TERMINATE_ON_FAILURE", "0") + .env( + "BT_EVAL_FILTER_PARSED", + r#"[{"path":["name"],"pattern":"^selected$"}]"#, + ) + .env_remove("BT_EVAL_SSE_SOCK") + .env_remove("BT_EVAL_SSE_ADDR") + .env_remove("BRAINTRUST_API_KEY") + .env("BRAINTRUST_APP_URL", "http://127.0.0.1:9") + .output() + .unwrap(); + let stdout = String::from_utf8(output.stdout).unwrap(); + assert!( + output.status.success(), + "{stdout}\n{}", + String::from_utf8_lossy(&output.stderr) + ); + assert!(!stdout.contains("Running evaluation:")); + assert!(!stdout.contains("Case 1:")); + if mode != "list" { + let lines: Vec<_> = stdout + .lines() + .filter(|line| line.starts_with('{')) + .collect(); + assert_eq!(lines.len(), 1, "{stdout}"); + let summary: serde_json::Value = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(summary["name"], "selected"); + assert_eq!(summary["total_cases"], 4); + assert!(summary.get("recording").is_none()); + } else { + assert!(stdout.lines().any(|line| line == "selected"), "{stdout}"); + assert!(!stdout.contains("excluded")); + assert!(!stdout.contains('{')); + } + } +} + +#[test] +fn source_and_reporting_failures_both_survive_in_partial_result() { + let mut opts = options(Arc::new(AtomicUsize::new(0))); + opts.source = Box::new(IterCaseSource::new( + vec![ + Ok(Case { + input: 3, + ..Default::default() + }), + Err(anyhow::anyhow!("source failed")), + ] + .into_iter(), + )); + let completion = run(EvalRequest::local(opts).with_reporter(FailedReporter { + stage: ReportStage::Finished, + attempts: Arc::new(AtomicUsize::new(0)), + })); + let error = futures::executor::block_on(async { completion.await }).unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [ + EvalRunFailure::Source { case_index: 1, .. }, + EvalRunFailure::Reporting { + stage: ReportStage::Finished, + .. + } + ] + )); + assert_eq!(error.partial.unwrap().results[0].output, Some(6)); +} +#[test] +fn dropping_the_observer_does_not_cancel_run_owned_reporting() { + struct Notice(Option>); + impl EvalReporter for Notice { + fn finished(&mut self, summary: &EvalSummary) -> Result<(), ReportError> { + self.0.take().unwrap().send(summary.total_cases).unwrap(); + Ok(()) + } + } + let (sender, receiver) = std::sync::mpsc::channel(); + let token = run(EvalRequest::local(options(Arc::new(AtomicUsize::new(0)))) + .with_reporter(Notice(Some(sender)))); + assert!(token.is_ok()); + drop(token); + assert_eq!(receiver.recv_timeout(Duration::from_secs(5)).unwrap(), 4); +} +#[test] +fn jsonl_write_and_flush_failures_remain_io_errors_with_computed_results() { + struct Broken { + fail_flush: bool, + } + impl Write for Broken { + fn write(&mut self, bytes: &[u8]) -> io::Result { + if self.fail_flush { + Ok(bytes.len()) + } else { + Err(io::ErrorKind::BrokenPipe.into()) + } + } + fn flush(&mut self) -> io::Result<()> { + Err(io::ErrorKind::BrokenPipe.into()) + } + } + for fail_flush in [false, true] { + let token = run(EvalRequest::local(options(Arc::new(AtomicUsize::new(0)))) + .with_reporter(JsonlReporter::new(Broken { fail_flush }))); + let error = futures::executor::block_on(async { token.await }).unwrap_err(); + assert!(matches!( + error.failures.as_slice(), + [EvalRunFailure::Reporting { + stage: ReportStage::Finished, + source: ReportError::Io(_), + .. + }] + )); + assert_eq!(error.partial.unwrap().successful_cases, 4); + } +} diff --git a/tests/eval_test.rs b/tests/eval_test.rs index 29ebdb2..765e937 100644 --- a/tests/eval_test.rs +++ b/tests/eval_test.rs @@ -859,7 +859,7 @@ async fn experiment_is_resolved_before_source_reads_and_case_roots_keep_independ } assert_eq!(roots.len(), 2); assert_eq!( - summary_to_value(&summary)["recording"]["experiment_id"], + summary_to_value(&summary).unwrap()["recording"]["experiment_id"], "chosen-id" ); }