feat: Turn batch evals API into API for evals deferred AI provider calls

.changeset/workflow-eval-submissions.md

@@ -0,0 +1,5 @@
+---
+"braintrust": minor
+---
+
+feat: Turn batch evals API into API for evals deferred AI provider calls

e2e/config/pr-comment-scenarios.json

@@ -684,12 +684,12 @@
     ]
   },
   {
-    "scenarioDirName": "durable-eval-webhook",
-    "label": "Durable Eval Webhook",
-    "metadataScenario": "durable-eval-webhook",
+    "scenarioDirName": "workflow-eval-webhook",
+    "label": "Workflow Eval Webhook",
+    "metadataScenario": "workflow-eval-webhook",
     "evals": [
       {
-        "experimentNameTemplate": "durable-eval-webhook-{testRunId}",
+        "experimentNameTemplate": "workflow-eval-webhook-{testRunId}",
         "label": "webhook"
       }
     ]

e2e/scenarios/durable-eval-webhook/scenario.test.ts → e2e/scenarios/workflow-eval-webhook/scenario.test.ts

@@ -10,7 +10,7 @@ const scenarioDir = await prepareScenarioDir({
   scenarioDir: resolveScenarioDir(import.meta.url),
 });
 
-test("durable eval collects task and scorer webhook sub-batches", async () => {
+test("workflow eval advances individual task and scorer submissions", async () => {
   await withScenarioHarness(
     async ({ events, runScenarioDir, testRunEvents }) => {
       await runScenarioDir({ scenarioDir });
@@ -30,17 +30,17 @@ test("durable eval collects task and scorer webhook sub-batches", async () => {
             JSON.stringify(left).localeCompare(JSON.stringify(right)),
           ),
       ).toEqual([
-        { batch_exact: 1, exact: 1 },
-        { batch_exact: 1, exact: 1 },
-        { batch_exact: 1, exact: 1 },
+        { workflow_exact: 1, exact: 1 },
+        { workflow_exact: 1, exact: 1 },
+        { workflow_exact: 1, exact: 1 },
+      ]);
+      expect(
+        webhookSpans.map((event) => event.metadata?.workflow_eval),
+      ).toEqual([
+        expect.objectContaining({ run_id: expect.any(String) }),
+        expect.objectContaining({ run_id: expect.any(String) }),
+        expect.objectContaining({ run_id: expect.any(String) }),
       ]);
-      expect(webhookSpans.map((event) => event.metadata?.durable_eval)).toEqual(
-        [
-          expect.objectContaining({ run_id: expect.any(String) }),
-          expect.objectContaining({ run_id: expect.any(String) }),
-          expect.objectContaining({ run_id: expect.any(String) }),
-        ],
-      );
 
       const taskSpans = findAllSpans(events(), "task");
       expect(taskSpans).toHaveLength(3);
@@ -59,18 +59,16 @@ test("durable eval collects task and scorer webhook sub-batches", async () => {
         "shared-eval-runtime",
       ]);
 
-      const batchScoreSpans = findAllSpans(events(), "batch_exact");
-      expect(batchScoreSpans).toHaveLength(3);
-      expect(batchScoreSpans.map((event) => event.scores)).toEqual([
-        { batch_exact: 1 },
-        { batch_exact: 1 },
-        { batch_exact: 1 },
-      ]);
-      expect(batchScoreSpans.map((event) => event.metadata?.method)).toEqual([
-        "batch-provider",
-        "batch-provider",
-        "batch-provider",
+      const workflowScoreSpans = findAllSpans(events(), "workflow_exact");
+      expect(workflowScoreSpans).toHaveLength(3);
+      expect(workflowScoreSpans.map((event) => event.scores)).toEqual([
+        { workflow_exact: 1 },
+        { workflow_exact: 1 },
+        { workflow_exact: 1 },
       ]);
+      expect(workflowScoreSpans.map((event) => event.metadata?.method)).toEqual(
+        ["workflow-provider", "workflow-provider", "workflow-provider"],
+      );
 
       const classifierSpans = findAllSpans(events(), "quality");
       expect(classifierSpans).toHaveLength(3);

e2e/scenarios/workflow-eval-webhook/scenario.ts

@@ -0,0 +1,163 @@
+import {
+  WorkflowScorer,
+  WorkflowTask,
+  defineWorkflowEval,
+  WorkflowEvalMemoryStore,
+} from "braintrust";
+import {
+  getTestRunId,
+  runMain,
+  scopedName,
+} from "../../helpers/scenario-runtime";
+
+async function main() {
+  const testRunId = getTestRunId();
+  const scenario = "workflow-eval-webhook";
+  const store = new WorkflowEvalMemoryStore();
+  const jobs = new Map<
+    string,
+    { input: number; output?: number; expected: number }
+  >();
+  const webhookCompletion = {
+    mode: "webhook" as const,
+    getExternalId: (submissionData: { id: string }) => submissionData.id,
+  };
+  const task = new WorkflowTask<
+    number,
+    number,
+    number,
+    { testRunId: string; kind: string },
+    Record<string, never>,
+    { id: string }
+  >({
+    async submit(item) {
+      const id = `task-${jobs.size + 1}`;
+      jobs.set(id, item);
+      return { id };
+    },
+    completion: webhookCompletion,
+    async collect(submissionData) {
+      return { output: jobs.get(submissionData.id)!.input * 2 };
+    },
+  });
+  const scorer = new WorkflowScorer<
+    number,
+    number,
+    number,
+    { testRunId: string; kind: string },
+    { id: string }
+  >({
+    name: "workflow_exact",
+    async submit(item) {
+      const id = `score-${jobs.size + 1}`;
+      jobs.set(id, item);
+      return { id };
+    },
+    completion: webhookCompletion,
+    async collect(submissionData) {
+      const item = jobs.get(submissionData.id)!;
+      return {
+        score: {
+          score: item.output === item.expected ? 1 : 0,
+          metadata: { method: "workflow-provider" },
+        },
+      };
+    },
+  });
+  const definition = defineWorkflowEval(
+    scopedName("e2e-workflow-eval-webhook-project", testRunId),
+    {
+      store,
+      maxConcurrency: 2,
+      experimentName: `${scenario}-${testRunId}`,
+      data: [1, 2, 3].map((input) => ({
+        id: `case-${input}`,
+        input,
+        expected: input * 2,
+        metadata: { scenario, testRunId, kind: "webhook" },
+      })),
+      task,
+      scores: [
+        function exact({ output, expected }) {
+          localScoreCount++;
+          return {
+            score: output === expected ? 1 : 0,
+            metadata: { method: "shared-eval-runtime" },
+          };
+        },
+        scorer,
+      ],
+      classifiers: [
+        function quality({ output, expected }) {
+          classifierCount++;
+          return {
+            name: "quality",
+            id: output === expected ? "pass" : "fail",
+            label: output === expected ? "Pass" : "Fail",
+          };
+        },
+      ],
+    },
+  );
+
+  let localScoreCount = 0;
+  let classifierCount = 0;
+  const waiting = await definition.start();
+  if (waiting.status !== "waiting" || jobs.size !== 3) {
+    throw new Error(
+      "Workflow eval did not pause with three webhook submissions",
+    );
+  }
+
+  // Finish the second case, including scoring, while the other tasks wait.
+  const taskIds = [...jobs.keys()];
+  const first = await definition.processSubmissionResult({
+    runId: waiting.runId,
+    externalId: taskIds[1],
+  });
+  if (
+    first.status !== "waiting" ||
+    first.pending.webhook !== 3 ||
+    localScoreCount !== 1 ||
+    classifierCount !== 1
+  ) {
+    throw new Error(
+      "Completed case did not advance its scorers and classifier independently",
+    );
+  }
+  const firstScoreId = [...jobs.keys()].find((id) => id.startsWith("score-"));
+  if (!firstScoreId)
+    throw new Error("Completed task did not submit its workflow scorer");
+  const scored = await definition.processSubmissionResult({
+    runId: waiting.runId,
+    externalId: firstScoreId,
+  });
+  if (scored.status !== "waiting" || scored.pending.webhook !== 2) {
+    throw new Error("Eval completed before the remaining tasks");
+  }
+  // A repeated delivery must not produce another scorer or log duplicate results.
+  await definition.processSubmissionResult({
+    runId: waiting.runId,
+    externalId: taskIds[1],
+  });
+  const completedJobs = new Set([taskIds[1], firstScoreId]);
+  for (const externalId of jobs.keys()) {
+    if (completedJobs.has(externalId)) continue;
+    await definition.processSubmissionResult({
+      runId: waiting.runId,
+      externalId,
+    });
+    completedJobs.add(externalId);
+  }
+  const completed = await definition.status({ runId: waiting.runId });
+  if (
+    completed.status !== "completed" ||
+    [...jobs.keys()].filter((id) => id.startsWith("score-")).length !== 3
+  ) {
+    throw new Error(
+      "Workflow eval did not complete three individual workflow scorers",
+    );
+  }
+}
+
+runMain(main);