Make evaluation execution reusable and reliable

examples/eval_async.rs

@@ -1,5 +1,4 @@
 use braintrust_sdk_rust::eval::*;
-use braintrust_sdk_rust::*;
 
 /// Demonstrates using BtEvalRunner, which is compatible with the `bt eval` CLI.
 ///
@@ -10,7 +9,7 @@ use braintrust_sdk_rust::*;
 ///   - BT_EVAL_SSE_SOCK=...  — send SSE events over a Unix socket
 ///   - BT_EVAL_SSE_ADDR=...  — send SSE events over TCP
 #[tokio::main]
-async fn main() -> Result<()> {
+async fn main() -> anyhow::Result<()> {
     // BtEvalRunner reads BT_EVAL_* env vars automatically.
     // In non-list mode it creates a BraintrustClient from the environment.
     let mut runner = BtEvalRunner::from_env().await?;
@@ -39,19 +38,35 @@ async fn main() -> Result<()> {
         },
     ];
 
+    // Share callbacks across runs; each run supplies its own source and options.
+    let definition = EvalDefinition::new(
+        Box::new(task(|input: &String, _hooks| {
+            Box::pin(async move { Ok(input.to_uppercase()) })
+        })),
+        vec![
+            Box::new(ExactMatch::new()),
+            Box::new(LevenshteinScorer::new()),
+            Box::new(scorer(
+                "nonempty",
+                |args: ScorerArgs<'_, String, String>| {
+                    Box::pin(async move {
+                        Ok(vec![Score::new(
+                            "nonempty",
+                            if args.output.is_empty() { 0.0 } else { 1.0 },
+                        )])
+                    })
+                },
+            )),
+        ],
+    )?;
+
     runner
         .eval(
             "my-project",
             EvalOpts::builder()
                 .name("uppercase-eval".to_string())
-                .dataset(cases.into())
-                .task(Box::new(SimpleFnTask::new(|input: &String| {
-                    Ok(input.to_uppercase())
-                })))
-                .scorers(vec![
-                    Box::new(ExactMatch::new()),
-                    Box::new(LevenshteinScorer::new()),
-                ])
+                .source(cases.into())
+                .definition(definition.clone())
                 .max_concurrency(2)
                 .build(),
         )

examples/eval_sync.rs

@@ -1,15 +1,11 @@
 use braintrust_sdk_rust::eval::*;
 use braintrust_sdk_rust::*;
+use std::future::IntoFuture;
 
 fn main() -> anyhow::Result<()> {
-    // Note: No #[tokio::main] needed! SyncEvaluator handles async internally
-
-    // Create Braintrust client
-    let rt = tokio::runtime::Runtime::new().map_err(|e| anyhow::anyhow!(e))?;
-    let client = rt.block_on(async { BraintrustClient::builder().build().await })?;
-
-    // Create sync evaluator
-    let evaluator = SyncEvaluator::new(client)?;
+    // 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![
@@ -39,13 +35,17 @@ fn main() -> anyhow::Result<()> {
     // Run evaluation synchronously
     println!("Starting synchronous evaluation...\n");
 
-    let result = evaluator.run(
-        EvalOpts::builder()
-            .name("sync-uppercase-eval".to_string())
-            .dataset(cases.into())
-            .task(Box::new(task))
-            .scorers(scorers)
-            .build(),
+    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(),
     )?;
 
     // Print results

src/eval/blocking.rs

@@ -0,0 +1,68 @@
+//! Managed blocking invocation on the run's fixed execution binding.
+use crate::internal::execution::ExecutionBinding;
+
+pub(crate) async fn invoke<T: Send + 'static>(
+    binding: Option<&ExecutionBinding>,
+    work: impl FnOnce() -> anyhow::Result<T> + Send + 'static,
+) -> anyhow::Result<T> {
+    let binding = binding.ok_or_else(|| {
+        anyhow::anyhow!("blocking callbacks require a managed evaluation executor")
+    })?;
+    let task = binding.spawn_blocking(work);
+    // Abort queued work on run cancellation. Tokio cannot interrupt a closure
+    // already running; its owned data remains valid until that closure returns.
+    struct AbortOnDrop(tokio::task::AbortHandle);
+    impl Drop for AbortOnDrop {
+        fn drop(&mut self) {
+            self.0.abort();
+        }
+    }
+    let _abort = AbortOnDrop(task.abort_handle());
+    task.await?
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::{
+        atomic::{AtomicBool, Ordering},
+        Arc,
+    };
+
+    #[test]
+    fn dropping_an_invocation_cancels_queued_work_before_it_can_start() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .max_blocking_threads(1)
+            .build()
+            .unwrap();
+        let binding = ExecutionBinding::external(runtime.handle().clone());
+        let (release, waiting) = std::sync::mpsc::channel();
+        let (entered, started) = std::sync::mpsc::channel();
+        let occupied = binding.spawn_blocking(move || {
+            entered.send(()).unwrap();
+            waiting
+                .recv_timeout(std::time::Duration::from_secs(3))
+                .unwrap()
+        });
+        started
+            .recv_timeout(std::time::Duration::from_secs(3))
+            .unwrap();
+        let called = Arc::new(AtomicBool::new(false));
+        runtime.block_on(async {
+            {
+                let called = called.clone();
+                let callback = invoke(Some(&binding), move || {
+                    called.store(true, Ordering::SeqCst);
+                    Ok(())
+                });
+                futures::pin_mut!(callback);
+                assert!(futures::poll!(callback).is_pending());
+            }
+            release.send(()).unwrap();
+            occupied.await.unwrap();
+            binding.spawn_blocking(|| ()).await.unwrap();
+        });
+        assert!(!called.load(Ordering::SeqCst));
+    }
+}

src/eval/bt_runner.rs

@@ -10,7 +10,7 @@
 use std::io::Write;
 use std::sync::{Arc, Mutex};
 
-use serde::{de::DeserializeOwned, Serialize};
+use serde::Serialize;
 use serde_json::{json, Value};
 
 use crate::{BraintrustClient, BraintrustError, Result};
@@ -186,20 +186,24 @@ fn create_sse_writer() -> Option<Arc<SseWriter>> {
 /// ```no_run
 /// use braintrust_sdk_rust::eval::*;
 ///
-/// # async fn example() -> braintrust_sdk_rust::Result<()> {
+/// # async fn example() -> anyhow::Result<()> {
 /// let mut runner = BtEvalRunner::from_env().await?;
 ///
+/// let definition = EvalDefinition::new(
+///     Box::new(SimpleFnTask::new(|input: &String| Ok(input.to_uppercase()))),
+///     vec![Box::new(ExactMatch::new())],
+/// )?;
+///
 /// runner.eval(
 ///     "my-project",
 ///     EvalOpts::builder()
 ///         .name("my-eval".to_string())
-///         .dataset(vec![Case {
+///         .source(vec![Case {
 ///             input: "hello".to_string(),
 ///             expected: Some("HELLO".to_string()),
 ///             ..Default::default()
 ///         }].into())
-///         .task(Box::new(SimpleFnTask::new(|input: &String| Ok(input.to_uppercase()))))
-///         .scorers(vec![Box::new(ExactMatch::new())])
+///         .definition(definition.clone())
 ///         .build(),
 /// ).await?;
 ///
@@ -288,8 +292,8 @@ impl BtEvalRunner {
     /// If the eval name does not pass the configured filters, it is skipped silently.
     pub async fn eval<I, O>(&mut self, project: &str, mut opts: EvalOpts<I, O>) -> Result<()>
     where
-        I: Serialize + DeserializeOwned + Send + Sync + Clone + 'static,
-        O: Serialize + DeserializeOwned + Send + Sync + PartialEq + Clone + 'static,
+        I: Serialize + Send + Sync + 'static,
+        O: Serialize + Send + Sync + 'static,
     {
         let name = opts.name.clone();
 
@@ -349,7 +353,7 @@ impl BtEvalRunner {
                     eprintln!("Evaluator '{}' failed: {}", name, e);
                 }
                 if self.config.terminate_on_failure {
-                    return Err(e);
+                    return Err(BraintrustError::Background(e.to_string()));
                 }
             }
         }