feat: Add batch evals api

[Luca Forstner (lforst)]

Adds an experimental workflow evaluation API for asynchronous provider operations that complete through polling or webhooks.
Each task or scorer submits one case/trial and collects one result, following the API in the JavaScript SDK PR.
As each task finishes, its scorers and classifiers can start while other tasks remain pending.

API

• define_workflow_eval(...) creates a WorkflowEval with start(), status(run_id), poll(run_id), and process_submission_result(run_id, ...).
• WorkflowTask.submit(item, context) receives one WorkflowTaskItem; collect(submission, context) returns one WorkflowTaskResult(output=..., metadata=..., tags=...).
• WorkflowScorer requires a name, submits one WorkflowScorerItem, and collects one WorkflowScorerResult(score=...).
• WorkflowSubmissionContext contains run_id and submission_id.
• Completion uses WorkflowSubmissionCompletionPoll or WorkflowSubmissionCompletionWebhook.
• max_concurrency bounds provider callbacks per invocation and defaults to 10.
• WorkflowEvalMemoryStore supports local runs; WorkflowEvalRedisStore persists state through an existing synchronous or asynchronous redis-py client.

Cases need stable IDs, either in the data or supplied by case_id.
Each trial gets its own item and submission IDs.
Submission data and collected values must be JSON serializable.
Task metadata merges into case metadata, and returned tags replace case tags.
Ordinary task functions, scorers, and classifiers can be mixed with workflow processors.

Define a workflow

The provider below is an application-owned adapter illustrating submission, status, and result retrieval.
Its methods should be implemented using the application's provider SDK.

from typing import TypedDict

from braintrust import (
    WorkflowEvalStore,
    WorkflowSubmissionCompletionPoll,
    WorkflowSubmissionCompletionWebhook,
    WorkflowSubmissionContext,
    WorkflowSubmissionPoll,
    WorkflowTask,
    WorkflowTaskItem,
    WorkflowTaskResult,
    define_workflow_eval,
)
from my_app import provider


class Submission(TypedDict):
    id: str


async def submit_task(
    item: WorkflowTaskItem[str, str],
    context: WorkflowSubmissionContext,
) -> Submission:
    request = await provider.submit(
        input=item.input,
        idempotency_key=context.submission_id,
        metadata={"braintrust_run_id": context.run_id},
    )
    return {"id": request.id}


async def collect_task(
    submission: Submission,
    _context: WorkflowSubmissionContext,
) -> WorkflowTaskResult[str]:
    output = await provider.result(submission["id"])
    return WorkflowTaskResult(output=output)


async def poll_submission(
    submission: Submission,
    _context: WorkflowSubmissionContext,
) -> WorkflowSubmissionPoll:
    request = await provider.retrieve(submission["id"])
    if request.status == "completed":
        return WorkflowSubmissionPoll("complete")
    if request.status == "failed":
        return WorkflowSubmissionPoll("failed", error=RuntimeError(request.error))
    return WorkflowSubmissionPoll("pending")


def make_workflow(
    store: WorkflowEvalStore,
    completion: (
        WorkflowSubmissionCompletionPoll[Submission]
        | WorkflowSubmissionCompletionWebhook[Submission]
    ),
):
    return define_workflow_eval(
        "tmp-luca-workflow-demo",
        store=store,
        data=[
            {"id": "france", "input": "Capital of France?", "expected": "Paris"},
            {"id": "japan", "input": "Capital of Japan?", "expected": "Tokyo"},
        ],
        task=WorkflowTask(
            submit=submit_task,
            completion=completion,
            collect=collect_task,
        ),
        scores=[lambda output, expected: output == expected],
        max_concurrency=10,
    )

A deferred scorer uses the same lifecycle:

from braintrust import WorkflowScorer, WorkflowScorerItem, WorkflowScorerResult


async def submit_score(
    item: WorkflowScorerItem[str, str, str],
    context: WorkflowSubmissionContext,
) -> Submission:
    request = await provider.submit_score(
        output=item.output,
        expected=item.expected,
        idempotency_key=context.submission_id,
        metadata={"braintrust_run_id": context.run_id},
    )
    return {"id": request.id}


async def collect_score(
    submission: Submission,
    _context: WorkflowSubmissionContext,
) -> WorkflowScorerResult:
    return WorkflowScorerResult(score=await provider.result(submission["id"]))


# Add this scorer to define_workflow_eval(..., scores=[...]).
scorer = WorkflowScorer(
    name="provider_judge",
    submit=submit_score,
    completion=WorkflowSubmissionCompletionPoll(poll=poll_submission),
    collect=collect_score,
)

Polling and webhooks

Use a shared store to resume across processes:

import os

from braintrust import WorkflowEvalRedisStore
from redis.asyncio import Redis


redis = Redis.from_url(os.environ["REDIS_URL"])
store = WorkflowEvalRedisStore(redis)
workflow = make_workflow(store, WorkflowSubmissionCompletionPoll(poll_submission))

# Inside an async worker; persist the run ID for subsequent invocations.
started = await workflow.start()
result = await workflow.poll(started.run_id)

Each poll() checks existing polling submissions once, without sleeping.
Newly submitted downstream work is checked on a later invocation.
status() reads progress without advancing work.
Waiting results include counts of pending polling and webhook submissions; completed results include the saved experiment summary.
Provider callback failures are raised after independent work advances, and polling can be retried.

For webhooks, construct the same definition with webhook completion:

workflow = make_workflow(
    store,
    WorkflowSubmissionCompletionWebhook(
        get_external_id=lambda submission, _context: submission["id"],
    ),
)
started = await workflow.start()

# In the application's handler, after verifying the provider webhook:
result = await workflow.process_submission_result(
    run_id,
    external_id=provider_request_id,
)
# The SDK's submission_id can also be used instead of external_id.

The application routes the event to its persisted run ID and handles provider failure events.
Collection callbacks must tolerate replay, including concurrent webhook delivery.
State records, external-ID lookups, atomic claims, and deduplicated progress sets support resumption and prevent duplicate downstream submissions.
Redis records and progress sets expire after seven days by default, configurable through ttl_ms.
Close the Redis client when the application shuts down.

Validation

• All 33 workflow tests pass, covering singular submission/collection, partial completion, concurrency limits, polling failures, webhook replay, fresh definitions, persistence recovery, metadata, tags, trials, and logging.
• nox -s test_types passes pyright, mypy, and all 22 runtime type tests.
• Focused pylint and all repository pre-commit hooks pass.
• Full test_core: 748 passed, 63 skipped, 12 xfailed, and four failures reproduced on the original PR commit (bb1283b8).
  Three local HTTP tests receive unexpected 502 responses; the git-metadata test has a cassette mismatch.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: 87a15e5e03

ℹ️ About Codex in GitHub

Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "Codex (@codex) review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "Codex (@codex) address that feedback".

[chatgpt-codex-connector]

Make action claims recoverable after callback interruption

When a submit, task, scorer, classifier, or finalization callback raises—or the worker stops after this claim but before writing its result—the claim remains permanently stored. Every subsequent poll() sees created=False, skips the incomplete action, and can never satisfy the missing stage record, leaving the run in waiting forever with the memory store and generally until the run itself expires with Redis. Use a recoverable lease/state transition or otherwise permit retrying claims whose corresponding result was not persisted.

Useful? React with 👍 / 👎.

[Luca Forstner (lforst)]

solving this right now is overkill