concurrent execution of evals

README.md

@@ -82,6 +82,8 @@ var eval = braintrust.<String, String>evalBuilder()
                                 (expected, result) -> expected.equals(result) ? 1.0 : 0.0))
                 .build();
 var result = eval.run();
+// TODO: document the concurrency contract - cases run 10-at-a-time by default;
+// task/scorers must be thread-safe. See Eval.Builder#maxConcurrency and #executor.
 System.out.println("\n\n" + result.createReportString());
 ```
 

braintrust-sdk/src/main/java/dev/braintrust/config/BraintrustConfig.java

@@ -64,6 +64,13 @@ public final class BraintrustConfig extends BaseConfig {
     /** Custom X509 trust manager for OTLP exporter. Builder-only field, not backed by envars. */
     private final X509TrustManager x509TrustManager;
 
+    /**
+     * Maximum number of eval cases the remote eval devserver evaluates concurrently. Matches {@link
+     * dev.braintrust.eval.Eval#DEFAULT_MAX_CONCURRENCY}.
+     */
+    private final int devserverMaxConcurrency =
+            getConfig("BRAINTRUST_DEVSERVER_MAX_CONCURRENCY", 10);
+
     /** CORS origins to allow when running remote eval devserver */
     private final String devserverCorsOriginWhitelistCsv =
             getConfig(
@@ -260,6 +267,12 @@ public Builder devserverCorsOriginWhitelistCsv(String csv) {
             return this;
         }
 
+        public Builder devserverMaxConcurrency(int maxConcurrency) {
+            envOverrides.put(
+                    "BRAINTRUST_DEVSERVER_MAX_CONCURRENCY", String.valueOf(maxConcurrency));
+            return this;
+        }
+
         public BraintrustConfig build() {
             return new BraintrustConfig(envOverrides, sslContext, x509TrustManager);
         }

braintrust-sdk/src/main/java/dev/braintrust/eval/Classifier.java

@@ -14,6 +14,8 @@
  * @param <INPUT> type of the input data
  * @param <OUTPUT> type of the output data
  */
+// TODO: document the concurrency contract - Eval invokes this from multiple threads
+// (see Eval.Builder#maxConcurrency), so implementations must be thread-safe.
 public interface Classifier<INPUT, OUTPUT> {
     String INVALID_CLASSIFICATION_MESSAGE =
             "When returning structured classifier results, each classification must be a non-empty"

braintrust-sdk/src/main/java/dev/braintrust/eval/ConcurrentCases.java

@@ -0,0 +1,87 @@
+package dev.braintrust.eval;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.Semaphore;
+import java.util.function.Consumer;
+import javax.annotation.Nullable;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Runs the cases of a dataset concurrently.
+ *
+ * <p>This is shared machinery behind {@link Eval} and the remote-eval devserver rather than an API
+ * for SDK users; it is public only because those two live in different packages.
+ */
+@Slf4j
+public final class ConcurrentCases {
+    private ConcurrentCases() {}
+
+    /**
+     * Drains {@code cursor} on the calling thread, running {@code caseConsumer} for each case on
+     * {@code executor}, with at most {@code maxConcurrency} cases in flight. Blocks until every
+     * case that was submitted has finished, then closes the cursor.
+     *
+     * <p>The drain is deliberately single-threaded: {@link Dataset.Cursor} is
+     * {@code @NotThreadSafe} and its {@code next()} may make network calls, so one thread pulls
+     * cases and fans them out. The calling thread only ever waits for permits — it never runs a
+     * case itself, so passing an {@code executor} that this thread belongs to cannot deadlock.
+     *
+     * <p>A throw from {@code caseConsumer} is contained to its own case and logged; it does not
+     * abort the remaining cases. Callers that need to record per-case outcomes should do so inside
+     * {@code caseConsumer}. This method does not propagate the caller's {@link
+     * io.opentelemetry.context.Context} onto worker threads — wrap {@code caseConsumer} if you need
+     * that.
+     *
+     * @return the error that aborted the drain (for example a failure fetching the next page of a
+     *     dataset), or null if every case was submitted
+     */
+    public static <CASE> @Nullable Throwable drain(
+            Dataset.Cursor<CASE> cursor,
+            Executor executor,
+            int maxConcurrency,
+            Consumer<CASE> caseConsumer) {
+        var inFlight = new Semaphore(maxConcurrency);
+        Throwable fatal = null;
+        try (cursor) {
+            for (var next = cursor.next(); next.isPresent(); next = cursor.next()) {
+                var item = next.get();
+                inFlight.acquire();
+                try {
+                    executor.execute(
+                            () -> {
+                                try {
+                                    caseConsumer.accept(item);
+                                } catch (Throwable t) {
+                                    // Contain the failure to this case: one bad case must not
+                                    // abort the rest of the run.
+                                    log.warn("Eval case failed", t);
+                                } finally {
+                                    inFlight.release();
+                                }
+                            });
+                } catch (RuntimeException e) {
+                    // e.g. RejectedExecutionException from a caller-supplied executor. Release the
+                    // permit we took so the drain below can't hang.
+                    inFlight.release();
+                    throw e;
+                }
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            fatal = e;
+        } catch (Throwable t) {
+            fatal = t;
+        } finally {
+            // Wait for every in-flight case, including when the drain above aborted.
+            try {
+                inFlight.acquire(maxConcurrency);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                if (fatal == null) {
+                    fatal = e;
+                }
+            }
+        }
+        return fatal;
+    }
+}

braintrust-sdk/src/main/java/dev/braintrust/eval/Dataset.java

@@ -32,6 +32,8 @@ default void forEach(Consumer<DatasetCase<INPUT, OUTPUT>> consumer) {
         }
     }
 
+    // TODO: document the concurrency contract - Eval drains a cursor from a single coordinator
+    // thread even when evaluating cases concurrently, so implementations need no locking.
     @NotThreadSafe
     interface Cursor<CASE> extends AutoCloseable {
         /**