From d2bee75f96fbdb8db6c31047dc147a1c93fc16e5 Mon Sep 17 00:00:00 2001 From: Stephen Belanger Date: Wed, 21 Oct 2026 17:06:50 +0800 Subject: [PATCH] Make CLI construction eager and consolidate the evaluation API --- examples/eval_async.rs | 3 +- examples/eval_sync.rs | 22 +- src/eval/bt_runner.rs | 272 +++++++++------------- src/eval/bt_runner/config.rs | 89 +++++++ src/eval/bt_runner/tests.rs | 112 ++++++--- src/eval/cli_session.rs | 25 ++ src/eval/mod.rs | 6 +- src/eval/{runner.rs => options.rs} | 49 +--- src/logger.rs | 51 ++-- tests/cli_initialization.rs | 360 +++++++++++++++++++++++++++++ tests/eval_callables.rs | 8 +- tests/eval_test.rs | 90 ++++---- 12 files changed, 765 insertions(+), 322 deletions(-) create mode 100644 src/eval/bt_runner/config.rs rename src/eval/{runner.rs => options.rs} (67%) create mode 100644 tests/cli_initialization.rs diff --git a/examples/eval_async.rs b/examples/eval_async.rs index 5e0c833..914526a 100644 --- a/examples/eval_async.rs +++ b/examples/eval_async.rs @@ -12,7 +12,8 @@ use braintrust_sdk_rust::eval::*; async fn main() -> anyhow::Result<()> { // BtEvalRunner reads BT_EVAL_* env vars automatically. // Recording mode creates a client; local and list modes need no credentials. - let mut runner = BtEvalRunner::from_env().await?; + let initialization = BtEvalRunner::from_env(); + let mut runner = initialization.await?; // Define test cases let cases = vec![ diff --git a/examples/eval_sync.rs b/examples/eval_sync.rs index 09e9c50..1931d54 100644 --- a/examples/eval_sync.rs +++ b/examples/eval_sync.rs @@ -5,7 +5,6 @@ use std::future::IntoFuture; fn main() -> anyhow::Result<()> { // The client's owned runtime drives work; this thread only waits for tokens. let client = futures::executor::block_on(BraintrustClient::builder().build().into_future())?; - let evaluator = Evaluator::new(client); // Define test cases let cases = vec![ @@ -36,17 +35,16 @@ fn main() -> anyhow::Result<()> { println!("Starting synchronous evaluation...\n"); let definition = EvalDefinition::new(Box::new(task), scorers)?; - let result = futures::executor::block_on( - evaluator - .run( - EvalOpts::builder() - .name("sync-uppercase-eval".to_string()) - .source(cases.into()) - .definition(definition.clone()) - .build(), - ) - .into_future(), - )?; + let completion = run(EvalRequest::recorded( + EvalOpts::builder() + .name("sync-uppercase-eval".to_string()) + .source(cases.into()) + .definition(definition.clone()) + .build(), + client, + )); + assert!(completion.is_ok()); + let result = futures::executor::block_on(completion.into_future())?; // Print results println!("\n=== Results ===\n"); diff --git a/src/eval/bt_runner.rs b/src/eval/bt_runner.rs index 3972051..0783972 100644 --- a/src/eval/bt_runner.rs +++ b/src/eval/bt_runner.rs @@ -13,99 +13,20 @@ use std::sync::{Arc, Mutex}; use serde::Serialize; use serde_json::{json, Value}; -use crate::{ - internal::execution::ExecutionBinding, BraintrustClient, BraintrustError, Completion, Result, -}; +use crate::{internal::execution::ExecutionBinding, BraintrustClient, BraintrustError, Completion}; use super::{ cli_session::{ready, Session}, - runner::EvalOpts, + options::EvalOpts, submission::{self, RunReporter}, types::EvalSummary, - BtEvalError, BtFinishError, EvalReporter, JsonlReporter, ReportError, + BtEvalError, BtFinishError, BtInitError, EvalReporter, JsonlReporter, ReportError, }; -/// A filter for selecting which evaluators to run, read from `BT_EVAL_FILTER_PARSED`. -struct EvalFilter { - path: Vec, - pattern: regex::Regex, -} - -impl EvalFilter { - /// Returns true if the given eval name matches this filter. - fn matches(&self, name: &str) -> bool { - // Match against the eval name when path is ["name"] or empty - if self.path.is_empty() || self.path == ["name"] { - return self.pattern.is_match(name); - } - // Unknown path — include by default - true - } -} - -/// Configuration read from `BT_EVAL_*` environment variables. -struct RunnerConfig { - /// Output summaries as JSONL to stdout (`BT_EVAL_JSONL=1`) - jsonl: bool, - /// List evaluator names without running them (`BT_EVAL_LIST=1`) - list: bool, - /// Stop running after the first failure (`BT_EVAL_TERMINATE_ON_FAILURE=1`) - terminate_on_failure: bool, - /// Filters to select which evaluators to run (`BT_EVAL_FILTER_PARSED`) - filters: Vec, - /// Skip sending logs to Braintrust (`BT_EVAL_NO_SEND_LOGS=1` or `BT_EVAL_LOCAL=1`) - no_send_logs: bool, -} - -impl RunnerConfig { - fn from_env() -> Self { - Self { - jsonl: env_flag("BT_EVAL_JSONL"), - list: env_flag("BT_EVAL_LIST"), - terminate_on_failure: env_flag("BT_EVAL_TERMINATE_ON_FAILURE"), - filters: parse_filters(), - no_send_logs: env_flag("BT_EVAL_NO_SEND_LOGS") || env_flag("BT_EVAL_LOCAL"), - } - } -} - -fn env_flag(name: &str) -> bool { - !matches!( - std::env::var(name).ok().as_deref(), - None | Some("") | Some("0") | Some("false") | Some("no") | Some("off") - ) -} - -fn parse_filters() -> Vec { - let serialized = match std::env::var("BT_EVAL_FILTER_PARSED").ok() { - Some(s) if !s.is_empty() => s, - _ => return vec![], - }; - - let parsed: Vec = match serde_json::from_str(&serialized) { - Ok(v) => v, - Err(e) => { - eprintln!("Invalid BT_EVAL_FILTER_PARSED: {}", e); - return vec![]; - } - }; - - parsed - .into_iter() - .filter_map(|v| { - let path = v["path"] - .as_array()? - .iter() - .filter_map(|s| s.as_str().map(String::from)) - .collect(); - let pattern_str = v["pattern"].as_str()?; - let pattern = regex::Regex::new(pattern_str) - .map_err(|e| eprintln!("Invalid filter pattern '{}': {}", pattern_str, e)) - .ok()?; - Some(EvalFilter { path, pattern }) - }) - .collect() -} +mod config; +#[cfg(test)] +use config::{env_flag, parse_filters, EvalFilter}; +use config::{RunnerConfig, SseEndpoint}; /// SSE event writer that sends events over a socket to the `bt eval` CLI. /// @@ -144,43 +65,47 @@ impl EvalReporter for SseReporter { } } -fn create_sse_writer() -> Option> { - // Unix socket SSE is only available on Unix platforms; Windows continues - // to the TCP fallback below. - #[cfg(unix)] - if let Ok(sock_path) = std::env::var("BT_EVAL_SSE_SOCK") { - if !sock_path.is_empty() { - match std::os::unix::net::UnixStream::connect(&sock_path) { - Ok(stream) => { - return Some(Arc::new(SseWriter { - writer: Mutex::new(Box::new(stream)), - })); - } - Err(e) => { - eprintln!("Failed to connect to SSE socket {}: {}", sock_path, e); - } - } +fn connect_sse(endpoint: SseEndpoint) -> std::result::Result, BtInitError> { + let writer: Box = match endpoint { + #[cfg(unix)] + SseEndpoint::Unix(path) => Box::new( + std::os::unix::net::UnixStream::connect(path).map_err(|source| BtInitError::Sse { + transport: "Unix socket", + source, + })?, + ), + SseEndpoint::Tcp(address) => { + let stream = + std::net::TcpStream::connect(address).map_err(|source| BtInitError::Sse { + transport: "TCP", + source, + })?; + let _ = stream.set_nodelay(true); + Box::new(stream) } - } - - // Try TCP address (BT_EVAL_SSE_ADDR) - if let Ok(addr) = std::env::var("BT_EVAL_SSE_ADDR") { - if !addr.is_empty() { - match std::net::TcpStream::connect(&addr) { - Ok(stream) => { - let _ = stream.set_nodelay(true); - return Some(Arc::new(SseWriter { - writer: Mutex::new(Box::new(stream)), - })); - } - Err(e) => { - eprintln!("Failed to connect to SSE address {}: {}", addr, e); - } - } + }; + Ok(Arc::new(SseWriter { + writer: Mutex::new(writer), + })) +} +async fn prepare_sse( + binding: &ExecutionBinding, + endpoint: Option, +) -> std::result::Result>, BtInitError> { + let Some(endpoint) = endpoint else { + return Ok(None); + }; + let work = binding.spawn_blocking(move || connect_sse(endpoint)); + struct Abort(tokio::task::AbortHandle); + impl Drop for Abort { + fn drop(&mut self) { + self.0.abort(); } } - - None + let _abort = Abort(work.abort_handle()); + work.await + .map_err(|_| BtInitError::Execution(crate::ExecutionError::WorkerLost))? + .map(Some) } /// A runner designed to work with the `bt eval` CLI. @@ -207,7 +132,8 @@ fn create_sse_writer() -> Option> { /// use braintrust_sdk_rust::eval::*; /// /// # async fn example() -> anyhow::Result<()> { -/// let mut runner = BtEvalRunner::from_env().await?; +/// let initialization = BtEvalRunner::from_env(); // initialization is admitted now +/// let mut runner = initialization.await?; /// /// let definition = EvalDefinition::new( /// Box::new(SimpleFnTask::new(|input: &String| Ok(input.to_uppercase()))), @@ -247,56 +173,82 @@ impl BtEvalRunner { /// Create a new runner, reading configuration from environment variables. /// /// 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(); - - 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 + /// `BRAINTRUST_API_KEY`). List and local modes create no client. Configuration + /// is captured and initialization admitted at the call; awaiting only observes + /// the result. Invalid filters reject immediately; SSE failures are returned + /// without switching output transports. Socket setup precedes client login. + pub fn from_env() -> Completion { + let config = match RunnerConfig::from_env() { + Ok(config) => config, + Err(error) => return Completion::rejected(error), }; + Self::initialize(config, None) + } - let binding = if let Some(client) = &client { - Some(client.execution_binding()) - } else if !config.list { - Some(ExecutionBinding::new(None)?) - } else { - None - }; + fn assemble( + config: RunnerConfig, + sse: Option>, + client: Option>, + binding: Option, + ) -> Self { let session = Session::new(config.terminate_on_failure); - Ok(Self { + Self { config, sse, client, binding, session, registered_names: Vec::new(), - }) + } } - /// Create a new runner with an explicit client. - /// - /// Environment variables are still read for mode configuration (list, JSONL, SSE, etc.), - /// but the provided client is used instead of creating a new one. All runs, - /// including local mode, inherit its fixed execution binding. - pub fn with_client(client: Arc) -> Self { - let config = RunnerConfig::from_env(); - let sse = create_sse_writer(); - let binding = Some(client.execution_binding()); - let session = Session::new(config.terminate_on_failure); - Self { - config, - sse, - client: Some(client), - binding, - session, - registered_names: Vec::new(), + /// Initialize now on this client's fixed binding. Dropping the completion + /// does not cancel socket setup. List mode uses neither its runtime nor SSE. + /// Configuration is captured from the environment at the call. + pub fn with_client(client: Arc) -> Completion { + let config = match RunnerConfig::from_env() { + Ok(config) => config, + Err(error) => return Completion::rejected(error), + }; + Self::initialize(config, Some(client)) + } + + fn initialize( + config: RunnerConfig, + supplied: Option>, + ) -> Completion { + if config.list { + return ready(Self::assemble(config, None, supplied, None)); + } + if let Some(client) = supplied { + let binding = client.execution_binding(); + let context = binding.clone(); + return binding.response(async move { + let sse = prepare_sse(&context, config.endpoint.clone()).await?; + Ok(Self::assemble(config, sse, Some(client), Some(context))) + }); + } + if config.no_send_logs { + let binding = match ExecutionBinding::new(None) { + Ok(binding) => binding, + Err(error) => return Completion::rejected(error.into()), + }; + let context = binding.clone(); + binding.response(async move { + let sse = prepare_sse(&context, config.endpoint.clone()).await?; + Ok(Self::assemble(config, sse, None, Some(context))) + }) + } else { + BraintrustClient::builder().build_with(|builder, binding| async move { + let sse = prepare_sse(&binding, config.endpoint.clone()).await?; + let client = builder.initialize(binding.clone()).await?; + Ok(Self::assemble( + config, + sse, + Some(Arc::new(client)), + Some(binding), + )) + }) } } diff --git a/src/eval/bt_runner/config.rs b/src/eval/bt_runner/config.rs new file mode 100644 index 0000000..bea9a8b --- /dev/null +++ b/src/eval/bt_runner/config.rs @@ -0,0 +1,89 @@ +//! Capture and validate CLI settings before admitting initialization. +use crate::eval::BtInitError; +use serde::Deserialize; + +pub(super) struct EvalFilter { + pub path: Vec, + pub pattern: regex::Regex, +} +impl EvalFilter { + pub fn matches(&self, name: &str) -> bool { + if self.path.is_empty() || self.path == ["name"] { + self.pattern.is_match(name) + } else { + true + } // Preserve existing handling of unsupported CLI filter paths. + } +} +#[derive(Clone)] +pub(super) enum SseEndpoint { + #[cfg(unix)] + Unix(std::path::PathBuf), + Tcp(String), +} +pub(super) struct RunnerConfig { + pub jsonl: bool, + pub list: bool, + pub terminate_on_failure: bool, + pub filters: Vec, + pub no_send_logs: bool, + pub endpoint: Option, +} +impl RunnerConfig { + pub fn from_env() -> Result { + let list = env_flag("BT_EVAL_LIST"); + // List output is stdout-only and requires no socket or runtime setup. + let mut endpoint = None; + if !list { + #[cfg(unix)] + if let Some(path) = std::env::var_os("BT_EVAL_SSE_SOCK").filter(|s| !s.is_empty()) { + endpoint = Some(SseEndpoint::Unix(path.into())); + } + if endpoint.is_none() { + if let Some(address) = std::env::var("BT_EVAL_SSE_ADDR") + .ok() + .filter(|s| !s.is_empty()) + { + endpoint = Some(SseEndpoint::Tcp(address)); + } + } + } + Ok(Self { + jsonl: env_flag("BT_EVAL_JSONL"), + list, + terminate_on_failure: env_flag("BT_EVAL_TERMINATE_ON_FAILURE"), + filters: parse_filters(std::env::var("BT_EVAL_FILTER_PARSED").ok().as_deref())?, + no_send_logs: env_flag("BT_EVAL_NO_SEND_LOGS") || env_flag("BT_EVAL_LOCAL"), + endpoint, + }) + } +} +pub(super) fn env_flag(name: &str) -> bool { + !matches!( + std::env::var(name).ok().as_deref(), + None | Some("") | Some("0") | Some("false") | Some("no") | Some("off") + ) +} +pub(super) fn parse_filters(serialized: Option<&str>) -> Result, BtInitError> { + let Some(serialized) = serialized.filter(|s| !s.is_empty()) else { + return Ok(vec![]); + }; + #[derive(Deserialize)] + struct Filter { + path: Vec, + pattern: String, + } + let filters: Vec = serde_json::from_str(serialized).map_err(BtInitError::Filters)?; + filters + .into_iter() + .enumerate() + .map(|(index, filter)| { + let pattern = regex::Regex::new(&filter.pattern) + .map_err(|source| BtInitError::FilterPattern { index, source })?; + Ok(EvalFilter { + path: filter.path, + pattern, + }) + }) + .collect() +} diff --git a/src/eval/bt_runner/tests.rs b/src/eval/bt_runner/tests.rs index 57ebb36..ccf8086 100644 --- a/src/eval/bt_runner/tests.rs +++ b/src/eval/bt_runner/tests.rs @@ -3,45 +3,28 @@ use std::sync::Mutex; use super::*; use crate::eval::EvalRequest; -// Serialize tests that mutate BT_EVAL_FILTER_PARSED to avoid race conditions. -static FILTER_ENV_MUTEX: Mutex<()> = Mutex::new(()); - #[test] fn test_env_flag() { - // Not set → false assert!(!env_flag("BT_TEST_NONEXISTENT_FLAG_XYZ")); } - #[test] fn test_parse_filters_empty() { - let _guard = FILTER_ENV_MUTEX.lock().unwrap(); - // When BT_EVAL_FILTER_PARSED is not set, filters should be empty - std::env::remove_var("BT_EVAL_FILTER_PARSED"); - let filters = parse_filters(); - assert!(filters.is_empty()); + assert!(parse_filters(None).unwrap().is_empty()); + assert!(parse_filters(Some("")).unwrap().is_empty()); } - #[test] fn test_parse_filters_valid() { - let _guard = FILTER_ENV_MUTEX.lock().unwrap(); - std::env::set_var( - "BT_EVAL_FILTER_PARSED", - r#"[{"path":["name"],"pattern":"foo.*"}]"#, - ); - let filters = parse_filters(); + let filters = parse_filters(Some(r#"[{"path":["name"],"pattern":"foo.*"}]"#)).unwrap(); assert_eq!(filters.len(), 1); assert!(filters[0].matches("foobar")); assert!(!filters[0].matches("bar")); - std::env::remove_var("BT_EVAL_FILTER_PARSED"); } - #[test] fn test_parse_filters_invalid_json() { - let _guard = FILTER_ENV_MUTEX.lock().unwrap(); - std::env::set_var("BT_EVAL_FILTER_PARSED", "not-json"); - let filters = parse_filters(); - assert!(filters.is_empty()); - std::env::remove_var("BT_EVAL_FILTER_PARSED"); + assert!(matches!( + parse_filters(Some("not-json")), + Err(BtInitError::Filters(_)) + )); } #[test] @@ -81,8 +64,17 @@ fn local_config(terminate_on_failure: bool) -> RunnerConfig { terminate_on_failure, filters: vec![], no_send_logs: true, + endpoint: None, } } +fn local_reported_runner(config: RunnerConfig, sse: Arc) -> BtEvalRunner { + BtEvalRunner::assemble( + config, + Some(sse), + None, + Some(ExecutionBinding::new(None).unwrap()), + ) +} fn local_options(fail: bool) -> EvalOpts { super::super::EvalOpts::builder() .name("cli".into()) @@ -115,9 +107,7 @@ async fn sse_protocol_is_the_only_reporter_and_case_failures_update_cli_status() 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(); + let mut runner = local_reported_runner(local_config(false), sse); assert!(runner.client.is_none()); runner.eval("project", local_options(false)).await.unwrap(); runner.eval("project", local_options(true)).await.unwrap(); @@ -139,9 +129,7 @@ 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(); + let mut runner = local_reported_runner(local_config(true), sse); assert!(runner.eval("project", local_options(true)).await.is_err()); assert!(!runner.all_passed()); assert!(!runner.finish().await.unwrap()); @@ -606,9 +594,7 @@ async fn finish_closes_sse_before_publication_even_when_its_observer_is_dropped( notice: Some(notice), })), }); - let mut runner = BtEvalRunner::from_config(local_config(false), Some(sse)) - .await - .unwrap(); + let mut runner = local_reported_runner(local_config(false), sse); drop(runner.eval("project", quiet_options(1))); let finish = runner.finish(); assert!(finish.is_ok()); @@ -641,3 +627,63 @@ async fn later_issuance_failure_does_not_skip_earlier_accepted_work() { )); assert!(!runner.finish().await.unwrap()); } + +#[tokio::test] +async fn constructor_uses_one_client_response_slot_and_rejects_before_socket_work() { + use crate::internal::response_tasks::MAX_RESPONSE_TASKS; + let client = BraintrustClient::builder() + .runtime(crate::current_runtime()) + .skip_login(true) + .build() + .await + .unwrap(); + let binding = client.execution_binding(); + let gate = Arc::new(tokio::sync::Semaphore::new(0)); + for _ in 0..MAX_RESPONSE_TASKS { + let gate = gate.clone(); + let token = binding.response::<(), BraintrustError, _>(async move { + gate.acquire().await.unwrap().forget(); + Ok(()) + }); + assert!(token.is_ok()); + drop(token); + } + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let mut config = local_config(false); + config.endpoint = Some(SseEndpoint::Tcp(listener.local_addr().unwrap().to_string())); + let rejected = BtEvalRunner::initialize(config, Some(Arc::new(client.clone()))); + assert!(rejected.is_err()); + assert!(matches!( + rejected.await, + Err(BtInitError::Admission(crate::EnqueueError::Full)) + )); + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock) + ); + gate.add_permits(MAX_RESPONSE_TASKS); + binding.responses_drained().await.unwrap(); + let token = BtEvalRunner::initialize(local_config(false), Some(Arc::new(client.clone()))); + assert!(token.is_ok()); + assert_eq!(binding.outstanding_responses(), 1); + let runner = token.await.unwrap(); + assert!(runner.finish().await.unwrap()); + client + .shutdown(std::time::Duration::from_secs(5)) + .await + .unwrap(); + let rejected = BtEvalRunner::initialize(local_config(false), Some(Arc::new(client.clone()))); + assert!(rejected.is_err()); + assert!(matches!( + rejected.await, + Err(BtInitError::Admission(crate::EnqueueError::Closed)) + )); + let mut list = local_config(false); + list.list = true; + let token = BtEvalRunner::initialize(list, Some(Arc::new(client))); + assert!(token.is_ok()); + let list = token.await.unwrap(); + assert!(list.binding.is_none()); + assert!(list.sse.is_none()); + assert!(list.finish().await.unwrap()); +} diff --git a/src/eval/cli_session.rs b/src/eval/cli_session.rs index 7aa3f18..feeb0d7 100644 --- a/src/eval/cli_session.rs +++ b/src/eval/cli_session.rs @@ -11,6 +11,31 @@ use std::{ }; use tokio::sync::oneshot; +/// Failure to configure or initialize a CLI session. +#[derive(Debug, thiserror::Error)] +pub enum BtInitError { + #[error(transparent)] + Client(#[from] crate::BraintrustError), + #[error(transparent)] + Execution(#[from] ExecutionError), + #[error(transparent)] + Admission(#[from] EnqueueError), + #[error("invalid BT_EVAL_FILTER_PARSED: {0}")] + Filters(#[source] serde_json::Error), + #[error("invalid pattern in CLI filter {index}: {source}")] + FilterPattern { + index: usize, + #[source] + source: regex::Error, + }, + #[error("failed to connect to CLI SSE {transport}: {source}")] + Sse { + transport: &'static str, + #[source] + source: std::io::Error, + }, +} + /// CLI run failure. Detailed computation/recording results remain available. pub enum BtEvalError { Run(EvalRunError), diff --git a/src/eval/mod.rs b/src/eval/mod.rs index dd4a750..fe890f6 100644 --- a/src/eval/mod.rs +++ b/src/eval/mod.rs @@ -56,27 +56,27 @@ mod cli_session; mod definition; mod error; mod executor; +mod options; mod recorder; mod reporter; mod run; -mod runner; mod scorer; mod source; mod submission; mod task; mod types; -pub use cli_session::{BtEvalError, BtFinishError}; +pub use cli_session::{BtEvalError, BtFinishError, BtInitError}; pub use definition::EvalDefinition; pub use error::{CaseError, EvalConfigError, EvalRunError, EvalRunFailure}; // Re-export public API pub use bt_runner::{summary_to_value, BtEvalRunner}; +pub use options::EvalOpts; 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}; pub use source::{ CaseSource, DatasetCaseSource, DatasetVersionRef, IterCaseSource, SourceError, diff --git a/src/eval/runner.rs b/src/eval/options.rs similarity index 67% rename from src/eval/runner.rs rename to src/eval/options.rs index a1afb66..c00b4b7 100644 --- a/src/eval/runner.rs +++ b/src/eval/options.rs @@ -1,13 +1,6 @@ -use serde::Serialize; use serde_json::{Map, Value}; -use std::marker::PhantomData; -use std::sync::Arc; -use crate::BraintrustClient; - -use super::{ - source::CaseSource, types::EvalSummary, EvalConfigError, EvalDefinition, EvalRunError, -}; +use super::{source::CaseSource, EvalConfigError, EvalDefinition}; /// Options for running an evaluation #[derive(bon::Builder)] @@ -47,46 +40,6 @@ impl EvalOpts { } } -/// Client-bound convenience facade for recorded evaluations. -pub struct Evaluator { - client: Arc, - _phantom: PhantomData<(I, O)>, -} - -impl Evaluator -where - I: Serialize + Send + Sync + 'static, - O: Serialize + Send + Sync + 'static, -{ - /// Create a new evaluator - pub fn new(client: BraintrustClient) -> Self { - Self { - client: Arc::new(client), - _phantom: PhantomData, - } - } - - /// Create a new evaluator from an Arc client (avoids cloning) - pub fn from_arc(client: Arc) -> Self { - Self { - client, - _phantom: PhantomData, - } - } - - /// Submit a recorded run immediately on the client binding. - /// Dropping the completion does not cancel accepted work. - pub fn run( - &self, - opts: EvalOpts, - ) -> crate::Completion, EvalRunError> { - super::run(super::EvalRequest::recorded( - opts, - self.client.as_ref().clone(), - )) - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/src/logger.rs b/src/logger.rs index bfea629..e67fdd2 100644 --- a/src/logger.rs +++ b/src/logger.rs @@ -340,29 +340,48 @@ impl BraintrustClientBuilder { /// Schedule client initialization immediately and observe its result. /// Dropping the completion does not cancel initialization or login. pub fn build(self) -> crate::Completion { - let validate = || -> Result<()> { - Url::parse(self.app_url.as_deref().unwrap_or(DEFAULT_APP_URL)) - .map_err(|_| BraintrustError::InvalidConfig("invalid app_url".into()))?; - Url::parse(self.api_url.as_deref().unwrap_or(DEFAULT_API_URL)) - .map_err(|_| BraintrustError::InvalidConfig("invalid api_url".into()))?; - if !self.skip_login && self.api_key.is_none() { - return Err(BraintrustError::InvalidConfig( - "API key required: set BRAINTRUST_API_KEY or call .api_key()".into(), - )); - } - Ok(()) - }; - let binding = validate().and_then(|()| ExecutionBinding::new(self.runtime.clone())); + self.build_with(|builder, binding| async move { builder.initialize(binding).await }) + } + + /// Compose initialization with an internal adapter in one response task on + /// the same validated binding. No nested build token or second runtime. + pub(crate) fn build_with(self, work: M) -> crate::Completion + where + T: Send + 'static, + E: From + + From + + From + + Send + + 'static, + M: FnOnce(Self, ExecutionBinding) -> F + Send + 'static, + F: std::future::Future> + Send + 'static, + { + let binding = self + .validate() + .and_then(|()| ExecutionBinding::new(self.runtime.clone())); match binding { Ok(binding) => { let context = binding.clone(); - binding.response(async move { self.initialize(context).await }) + binding.response(async move { work(self, context).await }) } - Err(error) => crate::Completion::rejected(error), + Err(error) => crate::Completion::rejected(E::from(error)), } } - async fn initialize(self, execution: ExecutionBinding) -> Result { + fn validate(&self) -> Result<()> { + Url::parse(self.app_url.as_deref().unwrap_or(DEFAULT_APP_URL)) + .map_err(|_| BraintrustError::InvalidConfig("invalid app_url".into()))?; + Url::parse(self.api_url.as_deref().unwrap_or(DEFAULT_API_URL)) + .map_err(|_| BraintrustError::InvalidConfig("invalid api_url".into()))?; + if !self.skip_login && self.api_key.is_none() { + return Err(BraintrustError::InvalidConfig( + "API key required: set BRAINTRUST_API_KEY or call .api_key()".into(), + )); + } + Ok(()) + } + + pub(crate) async fn initialize(self, execution: ExecutionBinding) -> Result { let app_url_str = self.app_url.unwrap_or_else(|| DEFAULT_APP_URL.into()); let api_url_str = self.api_url.unwrap_or_else(|| DEFAULT_API_URL.into()); diff --git a/tests/cli_initialization.rs b/tests/cli_initialization.rs new file mode 100644 index 0000000..1e70a16 --- /dev/null +++ b/tests/cli_initialization.rs @@ -0,0 +1,360 @@ +use braintrust_sdk_rust::{eval::*, BraintrustClient, ExecutionError}; +use std::{future::IntoFuture, io::Read, net::TcpListener, sync::Arc, time::Duration}; + +fn command(mode: &str) -> std::process::Command { + let mut command = std::process::Command::new(std::env::current_exe().unwrap()); + command + .args(["--exact", "initialization_child", "--nocapture"]) + .env("BT_INIT_CHILD", mode) + .env("BT_EVAL_LOCAL", "1") + .env("BT_EVAL_NO_SEND_LOGS", "0") + .env("BT_EVAL_LIST", "0") + .env("BT_EVAL_JSONL", "1") + .env("BT_EVAL_TERMINATE_ON_FAILURE", "0") + .env_remove("BT_EVAL_FILTER_PARSED") + .env_remove("BT_EVAL_SSE_SOCK") + .env_remove("BT_EVAL_SSE_ADDR") + .env_remove("BRAINTRUST_API_KEY") + .env_remove("BRAINTRUST_ORG_NAME"); + command +} +fn check(command: &mut std::process::Command) { + let output = command.output().unwrap(); + assert!( + output.status.success(), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); +} +fn options() -> EvalOpts { + EvalOpts::builder() + .name("init".into()) + .source( + vec![Case { + input: 1, + ..Default::default() + }] + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &usize| Ok(*input))), + vec![], + ) + .unwrap(), + ) + .quiet(false) + .build() +} +#[test] +fn initialization_child() { + let Ok(mode) = std::env::var("BT_INIT_CHILD") else { + return; + }; + match mode.as_str() { + "filters-json" | "filters-pattern" => { + let token = BtEvalRunner::from_env(); + assert!(token.is_err()); + let Err(error) = futures::executor::block_on(token.into_future()) else { + panic!("invalid filters must reject") + }; + if mode == "filters-json" { + assert!(matches!(error, BtInitError::Filters(_))); + } else { + assert!(matches!(error, BtInitError::FilterPattern { index: 0, .. })); + } + assert!(std::error::Error::source(&error).is_some()); + } + "list" => { + let token = BtEvalRunner::from_env(); + assert!(token.is_ok()); + let mut runner = futures::executor::block_on(token.into_future()).unwrap(); + assert!(runner.is_list_mode()); + drop(runner.eval("project", options())); + assert!(futures::executor::block_on(runner.finish().into_future()).unwrap()); + } + "sse-error" | "unix-error" => { + let token = BtEvalRunner::from_env(); + assert!(token.is_ok()); + let Err(error) = futures::executor::block_on(token.into_future()) else { + panic!("failed SSE must not fall back") + }; + match error { + BtInitError::Sse { transport, source } => { + assert_eq!( + transport, + if mode == "unix-error" { + "Unix socket" + } else { + "TCP" + } + ); + assert!(matches!( + source.kind(), + std::io::ErrorKind::NotFound | std::io::ErrorKind::ConnectionRefused + )); + } + other => panic!("unexpected error: {other:?}"), + } + } + "snapshot" | "lost" => { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + let client = runtime + .block_on( + BraintrustClient::builder() + .runtime(Some(runtime.handle().clone())) + .skip_login(true) + .build() + .into_future(), + ) + .unwrap(); + let token = BtEvalRunner::with_client(Arc::new(client)); + assert!(token.is_ok()); + if mode == "lost" { + drop(runtime); + let Err(error) = futures::executor::block_on(token.into_future()) else { + panic!("runtime loss must fail") + }; + assert!(matches!( + error, + BtInitError::Execution(ExecutionError::WorkerLost) + )); + } else { + // No initialization task can run until this runtime is driven. + std::env::set_var("BT_EVAL_LIST", "1"); + std::env::set_var("BT_EVAL_FILTER_PARSED", "invalid"); + std::env::set_var("BT_EVAL_SSE_ADDR", "127.0.0.1:9"); + std::env::set_var("BT_EVAL_LOCAL", "0"); + runtime.block_on(async { + let mut runner = token.await.unwrap(); + assert!(!runner.is_list_mode()); + assert!(runner.no_send_logs()); + runner.eval("project", options()).await.unwrap(); + assert!(runner.finish().await.unwrap()); + }); + } + } + "record" => { + let token = BtEvalRunner::from_env(); + assert!(token.is_ok()); + futures::executor::block_on(async { + let mut runner = token.await.unwrap(); + assert!(!runner.no_send_logs()); + runner.eval("project", options()).await.unwrap(); + assert!(runner.finish().await.unwrap()); + }); + } + "drop" => { + let client = futures::executor::block_on( + BraintrustClient::builder() + .skip_login(true) + .build() + .into_future(), + ) + .unwrap(); + let token = BtEvalRunner::with_client(Arc::new(client.clone())); + assert!(token.is_ok()); + drop(token); + futures::executor::block_on(client.shutdown(Duration::from_secs(5)).into_future()) + .unwrap(); + } + other => panic!("unknown mode {other}"), + } +} +#[test] +fn invalid_filters_fail_before_client_or_socket_setup() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + for (mode, filters) in [ + ("filters-json", "{"), + ("filters-json", r#"[{"path":[1],"pattern":"a"}]"#), + ("filters-pattern", r#"[{"path":["name"],"pattern":"("}]"#), + ] { + check( + command(mode) + .env("BT_EVAL_LOCAL", "0") + .env("BT_EVAL_FILTER_PARSED", filters) + .env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + ), + ); + } + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock) + ); +} +#[test] +fn list_initialization_ignores_sse_and_needs_no_credentials() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + check( + command("list") + .env("BT_EVAL_LIST", "1") + .env("BT_EVAL_LOCAL", "0") + .env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + ), + ); + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock) + ); +} +#[test] +fn constructor_configuration_is_captured_before_first_poll() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + check(command("snapshot").env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + )); + let (mut stream, _) = listener + .accept() + .expect("captured SSE endpoint must be used"); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut text = String::new(); + stream.read_to_string(&mut text).unwrap(); + assert!(text.contains("event: start\n")); + assert!(text.contains("event: summary\n")); + assert!(!text.contains("event: error\n")); +} +#[test] +fn dropped_constructor_observer_still_connects_and_shutdown_joins_cleanup() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + check(command("drop").env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + )); + let (mut stream, _) = listener + .accept() + .expect("accepted initialization must survive observer drop"); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut bytes = Vec::new(); + stream.read_to_end(&mut bytes).unwrap(); + assert!( + bytes.is_empty(), + "discarded runner closes its socket without running evals" + ); +} +#[test] +fn external_runtime_loss_before_first_poll_does_not_connect_elsewhere() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + check(command("lost").env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + )); + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock) + ); +} +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn sse_failure_precedes_recording_client_login() { + let server = wiremock::MockServer::start().await; + let closed = TcpListener::bind("127.0.0.1:0").unwrap(); + let address = closed.local_addr().unwrap(); + drop(closed); + check( + command("sse-error") + .env("BT_EVAL_LOCAL", "0") + .env("BRAINTRUST_API_KEY", "test-key") + .env("BRAINTRUST_APP_URL", server.uri()) + .env("BRAINTRUST_API_URL", server.uri()) + .env("BT_EVAL_SSE_ADDR", address.to_string()), + ); + assert!(server.received_requests().await.unwrap().is_empty()); +} +#[cfg(unix)] +#[test] +fn explicit_unix_failure_does_not_fall_back_to_tcp() { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + let missing = + std::path::PathBuf::from("/tmp").join(format!("bt-init-{}.sock", uuid::Uuid::new_v4())); + check(command("unix-error").env("BT_EVAL_SSE_SOCK", missing).env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + )); + assert!( + matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock) + ); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn recording_constructor_composes_login_and_evaluation_with_the_same_client() { + use serde_json::json; + 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 listener = TcpListener::bind("127.0.0.1:0").unwrap(); + listener.set_nonblocking(true).unwrap(); + check( + command("record") + .env("BT_EVAL_LOCAL", "0") + .env("BRAINTRUST_API_KEY", "test-key") + .env("BRAINTRUST_APP_URL", server.uri()) + .env("BRAINTRUST_API_URL", server.uri()) + .env( + "BT_EVAL_SSE_ADDR", + listener.local_addr().unwrap().to_string(), + ), + ); + let (mut stream, _) = listener.accept().unwrap(); + stream + .set_read_timeout(Some(Duration::from_secs(3))) + .unwrap(); + let mut text = String::new(); + stream.read_to_string(&mut text).unwrap(); + assert!(text.contains("event: start\n")); + assert!(text.contains("\"experiment_id\":\"experiment\"")); + let requests = server.received_requests().await.unwrap(); + assert_eq!( + requests + .iter() + .filter(|r| r.url.path() == "/api/apikey/login") + .count(), + 1 + ); + assert_eq!( + requests + .iter() + .filter(|r| r.url.path() == "/api/experiment/register") + .count(), + 1 + ); + assert!(requests.iter().any(|r| r.url.path() == "/logs3")); +} diff --git a/tests/eval_callables.rs b/tests/eval_callables.rs index 5ce314e..1fe35ac 100644 --- a/tests/eval_callables.rs +++ b/tests/eval_callables.rs @@ -1,8 +1,8 @@ use async_trait::async_trait; use braintrust_sdk_rust::{ eval::{ - scorer, task, Case, CaseSource, EvalConfigError, EvalDefinition, EvalOpts, Evaluator, - ExactMatch, Score, Scorer, ScorerArgs, Scores, SimpleFnTask, Task, TaskHooks, + run, scorer, task, Case, CaseSource, EvalConfigError, EvalDefinition, EvalOpts, + EvalRequest, ExactMatch, Score, Scorer, ScorerArgs, Scores, SimpleFnTask, Task, TaskHooks, }, BraintrustClient, }; @@ -156,7 +156,7 @@ async fn invalid_concurrency_and_names_fail_before_login_recording_or_source_acc .build() .await .unwrap(); - let evaluator = Evaluator::new(client); + let execute = |options| run(EvalRequest::recorded(options, client.clone())); let accessed = Arc::new(AtomicBool::new(false)); for (concurrency, names, expected) in [ (0, vec!["one"], EvalConfigError::ZeroConcurrency), @@ -199,7 +199,7 @@ async fn invalid_concurrency_and_names_fail_before_login_recording_or_source_acc .build(); assert_eq!(opts.validate(), Err(expected.clone())); assert!( - matches!(evaluator.run(opts).await, Err(error) if matches!(&error.failures[0], braintrust_sdk_rust::eval::EvalRunFailure::Config(actual) if actual == &expected) && error.partial.is_none()) + matches!(execute(opts).await, Err(error) if matches!(&error.failures[0], braintrust_sdk_rust::eval::EvalRunFailure::Config(actual) if actual == &expected) && error.partial.is_none()) ); } assert!(!accessed.load(Ordering::SeqCst)); diff --git a/tests/eval_test.rs b/tests/eval_test.rs index 765e937..87ac862 100644 --- a/tests/eval_test.rs +++ b/tests/eval_test.rs @@ -48,7 +48,7 @@ async fn recording_client() -> (BraintrustClient, wiremock::MockServer) { #[tokio::test] async fn test_basic_eval_flow() { let (client, server) = recording_client().await; - let evaluator = Evaluator::new(client); + let execute = |options| run(EvalRequest::recorded(options, client.clone())); let cases = vec![ Case { @@ -80,17 +80,16 @@ async fn test_basic_eval_flow() { }, ))]; - let result = evaluator - .run( - EvalOpts::builder() - .name("test-basic-eval".to_string()) - .source(cases.into()) - .definition(EvalDefinition::new(Box::new(callback), scorers).unwrap()) - .quiet(true) - .build(), - ) - .await - .expect("Evaluation failed"); + let result = execute( + EvalOpts::builder() + .name("test-basic-eval".to_string()) + .source(cases.into()) + .definition(EvalDefinition::new(Box::new(callback), scorers).unwrap()) + .quiet(true) + .build(), + ) + .await + .expect("Evaluation failed"); assert_eq!(result.total_cases, 2); assert_eq!(result.successful_cases, 2); @@ -203,38 +202,37 @@ fn test_eval_opts_builder() { #[tokio::test] async fn recorded_scorer_failure_keeps_valid_scores_and_finishes_the_case() { let (client, server) = recording_client().await; - let evaluator = Evaluator::new(client); + let execute = |options| run(EvalRequest::recorded(options, client.clone())); let scorers: Vec>> = vec![ Box::new(ExactMatch::new()), Box::new(SyncFnScorer::new("broken", |_| { anyhow::bail!("scoring failed") })), ]; - let result = evaluator - .run( - EvalOpts::builder() - .name("failed-scorer".into()) - .source( - vec![Case { - input: "a".to_owned(), - expected: Some("A".to_owned()), - id: Some("source-case".into()), - ..Default::default() - }] - .into(), - ) - .definition( - EvalDefinition::new( - Box::new(SimpleFnTask::new(|input: &String| Ok(input.to_uppercase()))), - scorers, - ) - .unwrap(), + let result = execute( + EvalOpts::builder() + .name("failed-scorer".into()) + .source( + vec![Case { + input: "a".to_owned(), + expected: Some("A".to_owned()), + id: Some("source-case".into()), + ..Default::default() + }] + .into(), + ) + .definition( + EvalDefinition::new( + Box::new(SimpleFnTask::new(|input: &String| Ok(input.to_uppercase()))), + scorers, ) - .quiet(true) - .build(), - ) - .await - .unwrap(); + .unwrap(), + ) + .quiet(true) + .build(), + ) + .await + .unwrap(); assert_eq!(result.successful_cases, 0); assert_eq!(result.failed_cases, 1); assert_eq!(result.score_stats[0].name, "exact_match"); @@ -356,7 +354,7 @@ async fn source_failure_drains_active_cases_preserves_provenance_and_finishes_re .max_concurrency(3) .quiet(true) .build(); - let run = tokio::spawn(async move { Evaluator::new(client).run(opts).await }); + let run = tokio::spawn(async move { run(EvalRequest::recorded(opts, client)).await }); tokio::time::timeout(std::time::Duration::from_secs(3), failed.notified()) .await .expect("pending source reads must allow active task progress"); @@ -445,7 +443,7 @@ async fn failure_before_the_first_case_returns_the_resolved_target_without_trace ) .quiet(true) .build(); - let error = Evaluator::new(client).run(opts).await.unwrap_err(); + let error = run(EvalRequest::recorded(opts, client)).await.unwrap_err(); assert!( matches!(&error.failures[0],EvalRunFailure::Source {case_index:0,source} if source.to_string()=="source unavailable") ); @@ -517,7 +515,7 @@ async fn recording_failure_stops_new_cases_but_drains_already_scheduled_cases() .max_concurrency(2) .quiet(true) .build(); - let run = tokio::spawn(async move { Evaluator::new(client).run(opts).await }); + let run = tokio::spawn(async move { run(EvalRequest::recorded(opts, client)).await }); tokio::time::timeout(std::time::Duration::from_secs(3), entered.notified()) .await .unwrap(); @@ -572,7 +570,7 @@ async fn reusable_definition_keeps_recorded_runs_and_sources_independent() { Arc, }; let (client, _server) = recording_client().await; - let evaluator = Evaluator::new(client); + let execute = |options| run(EvalRequest::recorded(options, client.clone())); let calls = Arc::new(AtomicUsize::new(0)); let definition = EvalDefinition::new( Box::new(task({ @@ -602,8 +600,8 @@ async fn reusable_definition_keeps_recorded_runs_and_sources_independent() { .build() }; let (first, second) = tokio::join!( - evaluator.run(options("first", 2, 4)), - evaluator.run(options("second", 3, 100)), + execute(options("first", 2, 4)), + execute(options("second", 3, 100)), ); let first = first.unwrap(); let second = second.unwrap(); @@ -616,7 +614,7 @@ async fn reusable_definition_keeps_recorded_runs_and_sources_independent() { assert_eq!(first.results[0].tags, Some(vec!["input-2".into()])); assert_eq!(second.results[0].tags, Some(vec!["input-3".into()])); assert_ne!(first.results[0].span_id, second.results[0].span_id); - let later = evaluator.run(options("later", 5, 10)).await.unwrap(); + let later = execute(options("later", 5, 10)).await.unwrap(); assert_eq!(later.total_cases, 1); assert_eq!(later.results[0].output, Some(10)); assert_eq!(calls.load(Ordering::SeqCst), 3); @@ -657,7 +655,9 @@ async fn serialization_failures_keep_outputs_and_finalize_recording() { ) .quiet(true) .build(); - let error = Evaluator::new(client).run(options).await.unwrap_err(); + let error = run(EvalRequest::recorded(options, client)) + .await + .unwrap_err(); let fields = error .failures .iter()