← all docs

Benchwright Developer Guide

Last modified September 10, 2026

How to build real things against the Benchwright API. For the formal endpoint reference see api.md; this document is the task-first companion.

Each chapter is a complete program. Copy, tweak, ship.

1. Install and authenticate

# Python
pip install benchwright

# CLI (statically linked, no runtime)
curl -fsSL https://benchwright.ai/install.sh | sh

(A TypeScript SDK is planned; today the API is Python + CLI + raw HTTP.)

Create a personal access token at https://benchwright.ai/manage (API Keys). The token is shown once, prefixed bw_live_pat_… (every token carries that prefix today; bw_test_pat_… is reserved for a future sandbox environment). Pick scopes that match what you're building; the safe starter set for a notebook is:

runs:read runs:write artifacts:read secrets:read billing:read registry:read

For benchmark authors add registry:publish. For CI bots add billing:write only if the bot should be able to top up. Note that tokens are minted in the web UI (or by benchwright token create against a browser session); a PAT can never mint, rotate, or revoke other PATs, by design.

export BW_PAT="bw_live_pat_..."

Every Python example below assumes:

import os
from benchwright import Benchwright
bw = Benchwright(api_key=os.environ["BW_PAT"])

Verify with a one-liner:

print(bw.billing.balance.get())
# {'object': 'balance',
#  'balance': {'amount_cents': 4215, 'currency': 'USD'},
#  'month_spent': {'amount_cents': 1230, 'currency': 'USD'}, ...}

(Or run benchwright doctor, which also checks scopes and that at least one provider secret is configured.)

2. Your first run

You'll need:

run = bw.runs.launch(
    request="Run MMLU Abstract Algebra on openai/gpt-4o-mini via OpenRouter",
)

# Watch it. The stream ends when the run is terminal.
for evt in bw.runs.stream(run.id):
    if evt.type in ("PhaseStarted", "PhaseCompleted"):
        print(f"{evt.type}: {evt.data.phase}")
    if evt.type == "MaterializeTaskResult":
        print(f"  task {evt.data.task_id}  score={evt.data.score}")

# Summary.
run = bw.runs.get(run.id)
tasks = bw.runs.tasks(run.id)
print(f"\n{tasks.summary.pass_rate:.1%} pass rate, "
      f"${run.cost.billed.amount_cents/100:.2f} billed")

# Open the report in your browser.
print(run.links.dashboard)

That's the full happy path. Everything else in this guide is variations on it.

3. Core concepts in 60 seconds

Three ways to launch a run:

Mode Body field When to use
Natural-language request: "..." exploring, or no catalog entry exists yet
Pinned benchmark (+ optional subset / impl) known catalog id — especially product/built benchmarks
Replay replay.from_run noise check, A/B a new model, same impl byte-for-byte

Replay inherits the parent run's authored artifacts and rewinds straight to Materialize, so it skips the LLM-driven authoring phases and is 3-10x cheaper than a natural-language launch. The rule of thumb is: first run is natural-language, everything after is a replay (or a pin when you only need the catalog entry, not a prior impl).

Pinned mode is benchmark alone — do not also send request (mutual exclusion). The launcher resolves the catalog row, seeds a synthetic request, and for product builds sets product_subset_id. Optional subset/impl further pin when you have them; for a prior run's exact scorer bytes prefer replay.from_run.

Two more launch fields shape how and where a run executes:

4. Run an existing benchmark against a model

Once a benchmark is in the registry, repeat runs against it should be anchored to a previous run so every launch scores with the same impl. Today that anchor is replay (or a catalog benchmark pin — see §3):

# Find the benchmark you want and its best prior run.
bm = bw.registry.benchmarks.find(name="BFCL v4")

# First run against a benchmark: natural language. The pipeline
# authors the scorer and registers the run in the catalog.
anchor = bw.runs.launch(
    request="Run BFCL v4 parallel function calling on qwen3-32b",
    sut={"model": "qwen/qwen3-32b", "provider": "openrouter"},
    limits={"run": {"cost": 15.0}},
    metadata={"purpose": "baseline", "model_family": "qwen3"},
)
bw.runs.wait(anchor.id)

# Every run after that: replay the anchor. Same tasks, same impl,
# no authoring cost. Override the model to compare another SUT.
run = bw.runs.launch(
    replay={"from_run": anchor.id},
    sut={"model": "qwen/qwen3-14b", "provider": "openrouter"},
)
bw.runs.wait(run.id)

The replay shares the anchor's impl, so the registry's same-impl comparison view picks both up automatically.

Reading results

The /tasks endpoint carries both the raw rows and a server-computed summary, so one call answers "how did it do?":

t = bw.runs.tasks(run.id)
print(t.summary)
# {
#   'total': 200, 'passed': 164, 'failed': 36, 'pass_rate': 0.82,
#   'duration_ms_p50': 744, 'duration_ms_p95': 1801,
# }

Filter for deeper inspection (the list rows are slim: score, error class, an output preview; the full input/expected/output lives behind .payload()):

t = bw.runs.tasks(run.id, passed=False)
for task in t.data[:5]:
    full = t.payload(task.id)
    print(task.task_id, "→", full.output[:120])

Running a standard benchmark via a third-party harness

You don't have to author a scorer to run a well-known benchmark. Set harness to a third-party harness and pass its task id and args under metadata.<harness>_harness; the pipeline drives that harness instead of Atomize. This is the fast path to the hundreds of benchmarks the harnesses already ship.

run = bw.runs.launch(
    request="Run MMLU abstract_algebra on qwen3-32b via Inspect",
    harness="inspect",
    sut={"model": "qwen/qwen3-32b", "provider": "openrouter"},
    metadata={
        "inspect_harness": {
            "task": "inspect_evals/mmlu_0_shot",
            "args": {"subjects": ["abstract_algebra"]},
        },
        "limit": 50,          # cap Materialize to the first 50 samples
    },
)
bw.runs.wait(run.id)
print(bw.runs.tasks(run.id).summary)

Harnesses: benchwright (default, our own Driver-authored harness), inspect (Inspect AI / inspect_evals), lmeval (lm-evaluation-harness), lighteval, harbor (terminal-bench). Naming an explicit third-party harness without a task is fine: the Find phase resolves the harness task from the request text. To pin it instead, inspect, lmeval and lighteval read metadata.<harness>_harness.task; harbor takes its dataset from metadata.harbor_harness.dataset, plus metadata.harbor_task to narrow that dataset to a single named task.

Reasoning models + non-CoT scorers. A harness multiple-choice task that caps generation for a non-chain-of-thought scorer (e.g. inspect_evals/mmlu_0_shot's max_non_cot_tokens) truncates a reasoning model mid-think, so the answer never reaches the scorer and the score reads near the random floor. For reasoning models, prefer a CoT variant, a larger token budget, or the benchwright harness (which captures the model's answer directly).

Choosing the compute backend

By default sandboxes run on your account's default backend. Pass compute="sandbox0" to route a single run to your self-hosted sandbox0 endpoint (configured under Settings → Compute) without changing your account default; "e2b" forces managed compute. The backend you ran on is recorded as run.backend.

run = bw.runs.launch(
    request="Run BFCL v4 parallel function calling on qwen3-32b",
    sut={"model": "qwen/qwen3-32b", "provider": "openrouter"},
    compute="sandbox0",
)

5. Iterating on failures

Three distinct iteration shapes. Pick by what you think went wrong.

5.1 Fork to fix a scorer or task_impl

The failures look like a code bug in how you compare output vs expected. Fork the run, describe the fix in natural language, and the pipeline rewinds to the Atomize phase with your description as extra context.

fix = bw.runs.fork(
    run.id,
    description=(
        "Parallel function calling failures are order-sensitive. "
        "Rewrite score() in task_impl.py to compare sets of "
        "{name, canonical_args} instead of ordered lists."
    ),
    from_phase="atomize",
    idempotency_key=f"fix-order-scorer-{run.id}",
)
bw.runs.wait(fix.id)

Forks inherit the request, model, secrets, and metadata. They produce a new impl.id because the code changed; the two runs show up side by side in the registry with a "derived from" edge.

The fork's impl lives in the registry like any other, so future replays of the fork keep scoring with the fixed code.

Not yet available: bw.impls.create_from_run(...) (freezing a run's authored impl as a standalone registry impl) is on the roadmap but the server endpoint doesn't exist yet; the SDK/CLI stubs return 404 / "not yet available". Anchor to the fork run via replay.from_run instead.

5.2 Replay against the same model

"Was this noise?" Replay the run: same tasks, same impl, fresh samples. Any score change is attributable to sampling noise, not scorer drift.

rerun = bw.runs.launch(
    replay={"from_run": run.id},
)
bw.runs.wait(rerun.id)

before = bw.runs.tasks(run.id).summary.pass_rate
after  = bw.runs.tasks(rerun.id).summary.pass_rate
print(f"noise check: {before:.1%} → {after:.1%}")

If the two pass rates track closely, the failures are deterministic and you have a real capability gap; if they diverge a lot, the benchmark is noisy at this sample count.

A replay re-runs the parent's whole materialize scope. replay.only_failed / replay.task_ids (re-running just a slice) are documented in the API reference as planned fields but are not honored yet; the CLI rejects the equivalent flags on replays for the same reason.

5.3 Replay against a different model

"Is this a model problem?" Same tasks and scorer, different model.

bigger = bw.runs.launch(
    replay={"from_run": run.id},
    sut={"model": "qwen/qwen3-235b-a22b-thinking", "provider": "openrouter"},
)
bw.runs.wait(bigger.id)

bt = bw.runs.tasks(bigger.id).summary
print(f"bigger model: {bt.pass_rate:.1%}")

The new run shares the parent's impl.id, so the registry treats the two as directly comparable. The comparison view at /registry/{benchmark} picks this up automatically.

5.4 When to do which

Symptom Use
Failures have a systematic pattern in output Fork (§5.1)
Failures look stochastic across reruns Replay same model (§5.2)
Model clearly lacks the capability Replay bigger model (§5.3)

These compose. A typical iteration is: fork to fix the scorer → replay the fork against every model you care about → get an honest post-fix comparison without re-authoring anything.

6. Authoring a benchmark from scratch

Pull tasks from wherever, write your own scorer, ship. This is the workflow that makes Benchwright a benchmark operating system rather than just a runner.

Availability. The authoring surface works end to end: creating benchmark catalog entries (§6.1), uploading task rows (subsets.create / append_tasks), registering raw-code impls (impls.create), running them, and publishing runs. A subset needs a grader to be runnable by the pipeline rather than catalog-only (see §8.6 of the API reference); the server accepts one on subset create, but the Python SDK doesn't expose the parameter yet, so send it with benchwright bench subset create --grader-file grader.json or a raw POST /v1/benchmarks/{id}/subsets. Still missing: DELETE for benchmarks, subsets and impls, reading impl code back via GET /v1/impls/{id}/code, and impls.create_from_run (§5.1).

6.1 Scaffold the benchmark + subset

bm = bw.benchmarks.create(
    name="BFCL v4 (my mirror)",
    description="Berkeley Function Calling Leaderboard v4, mirrored locally.",
    category="tool-use",
    tags=["function-calling", "parallel"],
)
print(bm.id)   # bm_<uuid>

The benchmark starts visibility="private"; only your own runs can see it until a public run flips it (visibility is derived from the runs, see §6.5).

6.2 Load task rows

Anywhere you can produce {task_id, input, expected} dicts is a valid source. Here's a Hugging Face example:

from datasets import load_dataset

ds = load_dataset("gorilla-llm/Berkeley-Function-Calling-Leaderboard",
                  split="parallel_function_v4")
tasks = [
    {
        "task_id": row["id"],
        "input": {
            "question":  row["question"],
            "functions": row["function"],
        },
        "expected": row["ground_truth"],
    }
    for row in ds
]

sub = bw.benchmarks.subsets.create(
    benchmark=bm.id,
    slug="parallel-function-v4",
    name="Parallel Function v4",
    description="200 parallel function-calling questions from BFCL v4.",
    tasks=tasks,                   # capped at 1000 rows per request; use append_tasks beyond that
)

6.3 Write the scorer

task_impl.py implements three methods on a BenchTask subclass. extract reads one row from the dataset file the pipeline provides; solve calls the model through the proxy (the real provider key is injected server-side — the scorer only ever sees a fake key); score returns a float in [0, 1].

The scorer reaches the model at BW_MODEL_PROXY_URL (an OpenAI-compatible endpoint), names the model with BW_MODEL_MODEL, and sends Authorization: Bearer $BW_MODEL_PROXY_KEY when that key is set.

import textwrap

code = textwrap.dedent('''
    import json, os, requests
    from bench_task import BenchTask

    class Task(BenchTask):
        def extract(self, path, index):
            with open(path) as f:
                rows = [json.loads(line) for line in f]
            row = rows[index]
            return {
                "task_id": row["task_id"],
                "input":   row["input"],
                "expected": row["expected"],
            }

        def solve(self, index, input):
            headers = {}
            key = os.environ.get("BW_MODEL_PROXY_KEY")
            if key: headers["Authorization"] = f"Bearer {key}"
            r = requests.post(
                f"{os.environ['BW_MODEL_PROXY_URL']}/v1/chat/completions",
                headers=headers,
                json={
                    "model":    os.environ["BW_MODEL_MODEL"],
                    "messages": [{"role": "user", "content": input["question"]}],
                    "tools":    input["functions"],
                    "tool_choice": "auto",
                },
                timeout=120,
            )
            r.raise_for_status()
            calls = r.json()["choices"][0]["message"].get("tool_calls", [])
            return json.dumps([
                {"name": c["function"]["name"],
                 "args": json.loads(c["function"]["arguments"])}
                for c in calls
            ])

        def score(self, index, output, expected):
            got  = {(c["name"], json.dumps(c["args"], sort_keys=True))
                    for c in json.loads(output)}
            want = {(c["name"], json.dumps(c["args"], sort_keys=True))
                    for c in expected}
            return 1.0 if got == want else 0.0
''').strip()

impl = bw.impls.create(
    benchmark=bm.id,
    subset=sub.id,
    label="strict-set scorer v1",
    language="python",
    code=code,
    description="Compares tool-call sets ignoring order.",
)

6.4 Run it

run = bw.runs.launch(
    benchmark=bm.id,
    subset=sub.id,
    impl=impl.id,
    sut={"model": "qwen/qwen3-32b", "provider": "openrouter"},
    limits={"run": {"cost": 15.0}},
)
bw.runs.wait(run.id)
print(bw.runs.tasks(run.id).summary)

6.5 Publish

Visibility is run-derived: publishing your reference run is what makes the benchmark public in the registry.

bw.runs.update(run.id, visibility="public")   # the reference run
# or, CLI: benchwright run publish <run_id>

A public benchmark shows up in the registry at https://benchwright.ai/registry/{bm.id}. The reference run double-duties as the leaderboard seed. (Benchmark/subset rows don't take a direct visibility patch; the registry derives their visibility from the runs that reference them.)

6.6 Fixing a published benchmark

Benchmarks are wrong sometimes, and usually you find out after publishing: a mint step whose pattern is too narrow, a run_timeout shorter than the product's own boot, a judge rubric rewarding the wrong thing, a sample pointing at the wrong frame. For a BUILT (product) benchmark you own, both halves are editable in place:

benchwright bench recipe show my-bench                       # read the current recipe
benchwright bench recipe set my-bench --set run_timeout=900 --set trials=3 \
  --note 'product boots slower than 240s'
benchwright bench samples set my-bench --upsert-file fix.jsonl \
  --note 'q2 pointed at the wrong frame'

The same thing over HTTP, which is how an agent does it (an operator holding your PAT acts as you, so it can fix a benchmark it diagnoses without a human touching the database):

curl -X PATCH https://benchwright.ai/v1/build/subsets/my-bench/recipe \
  -H "Authorization: Bearer $BW_PAT" -H 'Content-Type: application/json' \
  -d '{"config":{"run_timeout":900,"trials":3},"note":"why"}'

Editing scoring is owner-only. Both PATCHes are gated on the benchmark's owner (admins too) and need registry:publish; anyone else gets a 404. Reading the definition back needs registry:read, so a read-only token can diff a recipe and propose a fix without being able to apply it.

Neither edit rewrites history, which is what makes editing a published benchmark safe:

You change What moves
the recipe The subset id stays (it addresses the samples, which didn't change). The scorer version moves: the recipe is content-addressed into task_impl.py, so the next run records a new impl id, and every run before the edit stays stamped with the scorer that actually graded it.
the samples A subset id is a hash of its tasks, so the edited set forks a new content-addressed subset carrying the recipe forward; the old subset is retired and stops resolving for new runs. Nothing that already ran is reinterpreted.

So a leaderboard never silently mixes two scorers or two sample sets: the ids say which was which. Full request/response shapes are in the API reference under "Editing a BUILT benchmark".

This applies to benchmarks built here. A benchmark that wraps an external harness is a different case: matching the official implementation is the whole point of the number, so those are reported as-is rather than corrected.

7. Distilling a hard subset from a sweep

Take a popular public benchmark, run a frontier panel against it, publish the universally-hard tasks as a new subset. The result is a purpose-built regression set.

Availability. Step 4 is the one gap: bw.impls.create_from_run(...) has no server endpoint yet and 404s today (§5.1). Until it lands, register the scorer explicitly with bw.impls.create(subset=..., label=..., code=...), passing the task_impl.py text you want the new subset scored with. Every other step here works.

from collections import Counter

# 1. Run the panel: one natural-language anchor authors the scorer,
#    then replays fan the same impl out across the panel.
panel = [
    "openai/gpt-4o-mini",
    "anthropic/claude-haiku-4-5",
    "google/gemini-2.5-flash",
    "qwen/qwen3-32b",
    "meta-llama/llama-3.3-70b-instruct",
]
anchor = bw.runs.launch(
    request="Run MMLU Pro (full) on " + panel[0],
    sut={"model": panel[0], "provider": "openrouter"},
    metadata={"sweep": "hard-mmlu-2026-04", "model": panel[0]},
    idempotency_key=f"hard-mmlu-{panel[0]}",
)
bw.runs.wait(anchor.id)
runs = [anchor] + [
    bw.runs.launch(
        replay={"from_run": anchor.id},
        sut={"model": m, "provider": "openrouter"},
        metadata={"sweep": "hard-mmlu-2026-04", "model": m},
        idempotency_key=f"hard-mmlu-{m}",
    )
    for m in panel[1:]
]
for r in runs: bw.runs.wait(r.id)

# 2. Bucket failures across the panel.
fail_counts = Counter()
payloads = {}
for r in runs:
    t = bw.runs.tasks(r.id, passed=False)
    for row in t.data:
        fail_counts[row.task_id] += 1
        if row.task_id not in payloads:
            p = t.payload(row.id)
            payloads[row.task_id] = {
                "task_id": row.task_id,
                "input":   p.input,
                "expected": p.expected,
            }
hard = [payloads[tid] for tid, n in fail_counts.items() if n == len(panel)]
print(f"{len(hard)} universally hard tasks")

# Keep the sweep manifest: run ids are the durable correlation
# handle (launch metadata is behavioral config, not a queryable tag).
import json
json.dump({"sweep": "hard-mmlu-2026-04", "runs": [r.id for r in runs]},
          open("sweep-manifest.json", "w"))

# 3. Scaffold the new benchmark + subset.
gauntlet = bw.benchmarks.create(
    name="MMLU Pro Frontier Gauntlet",
    description="MMLU Pro tasks that every model in the 2026-04 frontier panel failed.",
    category="reasoning",
    tags=["curated", "hard", "regression"],
)
sub = bw.benchmarks.subsets.create(
    benchmark=gauntlet.id,
    slug="v1-2026-04",
    name="Gauntlet v1 (April 2026)",
    description=f"Distilled from a {len(panel)}-model panel on MMLU Pro full.",
    tasks=hard,
)

# 4. Inherit the scorer from the parent runs (they all used the same impl).
impl = bw.impls.create_from_run(
    from_run=runs[0].id,
    benchmark=gauntlet.id,
    subset=sub.id,
    label="inherited from mmlu-pro/official",
)

# 5. Publish: run the gauntlet once and flip that run public
#    (visibility is run-derived, §6.5).
ref = bw.runs.launch(request="Run the MMLU Pro Frontier Gauntlet on qwen3-32b",
                     sut={"model": "qwen/qwen3-32b", "provider": "openrouter"})
bw.runs.wait(ref.id)
bw.runs.update(ref.id, visibility="public")
print(f"live at https://benchwright.ai/registry/{gauntlet.id}")

Pattern: the new benchmark is derivative (name its parents in the description and the per-task source links), but the registry treats it as first-class, so it gets its own leaderboard and can accept any model.

8. Composite benchmarks across multiple sources

§7 distills hard tasks from one benchmark. The natural next step is a meta-suite pulling hard tasks from many benchmarks into one subset, so a single run produces a capability index across domains. Same workflow as §7, with two additions:

The result: one run, many capability areas, one aggregate score, plus per-kind breakdowns surfaced by the registry.

8.1 The composite task shape

MMLU Pro rows look nothing like BFCL rows, which look nothing like GSM8K or HumanEval rows. The composite subset normalizes each row to a shape the scorer can route on:

{
  "task_id": "bm_mmlu_pro::mmlu_pro_01234",
  "input":    { "kind": "multiple_choice", "question": "...", "choices": ["A","B","C","D"] },
  "expected": { "kind": "multiple_choice", "value": "D" },
  "source":   { "benchmark": "bm_mmlu_pro", "subset": "sub_full", "task_id": "mmlu_pro_01234" }
}

input.kind and expected.kind are always the same. The duplication is deliberate: the BenchTask protocol hands solve() the input and score() the expected separately, so carrying the dispatch key on both keeps each method self-contained.

Four kinds cover most public benchmarks:

kind input shape expected.value shape
multiple_choice {question, choices} string: the correct letter
function_call {question, functions} (OpenAI tool shape) list of {name, args} dicts
math {question} string: the numeric answer
code {prompt} {test_code: "assert foo(1) == 2"}

Extending to new kinds (agentic trajectories, multi-turn chat, image-in) is adding a branch to the dispatcher; the catalog shape doesn't change.

8.2 Harvest failures across source sweeps

Assumes you've already run a frontier panel against each source benchmark (the §7 pattern) and tagged them with a shared sweep marker. Each panel produces its own benchmark→universally-hard list; we merge them into one bag.

import json
from collections import Counter

SOURCES = [
    {"bm": "MMLU Pro",   "kind": "multiple_choice"},
    {"bm": "BFCL v4",    "kind": "function_call"},
    {"bm": "GSM8K",      "kind": "math"},
    {"bm": "HumanEval",  "kind": "code"},
]

def shape_input(kind, raw_input):
    """Normalize a source benchmark's input into the composite shape."""
    if kind == "multiple_choice":
        return {"kind": kind,
                "question": raw_input["question"],
                "choices":  raw_input["choices"]}
    if kind == "function_call":
        return {"kind": kind,
                "question":  raw_input["question"],
                "functions": raw_input["functions"]}
    if kind == "math":
        return {"kind": kind,
                "question": raw_input.get("question") or raw_input["problem"]}
    if kind == "code":
        return {"kind": kind,
                "prompt": raw_input["prompt"]}
    raise ValueError(kind)

def shape_expected(kind, raw_expected, raw_input):
    if kind == "code":
        # HumanEval stores the test harness separately.
        return {"kind": kind, "test_code": raw_input["test"]}
    return {"kind": kind, "value": raw_expected}

gauntlet_tasks = []
per_source = {}

manifest = json.load(open("sweep-manifest.json"))   # run ids per §7

for src in SOURCES:
    # Panel discovery: run ids from the sweep manifest, narrowed to
        # this source benchmark. (Run ids are the correlation handle;
    # see the §16 note on metadata.)
    panel = [r for rid in manifest["runs"]
             for r in [bw.runs.get(rid)]
             if r.status == "ok" and r.benchmark_name == src["bm"]]
    if not panel:
        print(f"  {src['bm']:<16} skipped (no panel runs)")
        continue

    # Bucket failures across the panel. A task is universally hard
    # when every model failed it.
    fail_counts = Counter()
    payloads    = {}
    subset_id   = panel[0].get("subset_id")   # same subset across the panel
    for r in panel:
        tl = bw.runs.tasks(r.id, passed=False)
        for t in tl.data:
            fail_counts[t.task_id] += 1
            if t.task_id not in payloads:
                p = tl.payload(t.id)
                payloads[t.task_id] = {"input": p.input, "expected": p.expected}

    hard = [tid for tid, n in fail_counts.items() if n == len(panel)]
    per_source[src["bm"]] = len(hard)

    for tid in hard:
        p = payloads[tid]
        gauntlet_tasks.append({
            "task_id":  f"{src['bm']}::{tid}",
            "input":    shape_input(src["kind"], p["input"]),
            "expected": shape_expected(src["kind"], p["expected"], p["input"]),
            "source": {
                "benchmark": src["bm"],
                "subset":    subset_id,
                "task_id":   tid,
            },
        })
    print(f"  {src['bm']:<16} {len(hard)} universally-hard")

print(f"\ntotal: {len(gauntlet_tasks)} tasks, {len(per_source)} source benchmarks")

Namespacing task_id with {source_bm}::{original_id} keeps the composite's task_id uniqueness constraint satisfied even when two source benchmarks happen to use the same original id.

8.3 Scaffold the composite

bm = bw.benchmarks.create(
    name="Reasoning Gauntlet",
    description=(
        "Universally-hard tasks distilled from MMLU Pro, BFCL v4, "
        "GSM8K, and HumanEval. A single scorer dispatches on task kind "
        "so the aggregate pass rate is a cross-domain capability index."
    ),
    category="composite",
    tags=["curated", "hard", "multi-domain", "gauntlet"],
)

sub = bw.benchmarks.subsets.create(
    benchmark=bm.id,
    slug="v1-2026-04",
    name="Gauntlet v1 (April 2026)",
    description=(
        f"{sum(per_source.values())} tasks across "
        f"{len(per_source)} source benchmarks: "
        + ", ".join(f"{k}({v})" for k, v in per_source.items())
    ),
    tasks=gauntlet_tasks,
)

8.4 The dispatching scorer

One task_impl.py. solve() and score() both read kind and route to the appropriate strategy. Adding a new kind means adding two branches and an optional helper.

import textwrap

code = textwrap.dedent('''
    import json, os, re, subprocess, tempfile, requests
    from bench_task import BenchTask

    def _chat(messages, **kwargs):
        """Single helper for every kind. Auth is injected server-side."""
        headers = {}
        key = os.environ.get("BW_MODEL_PROXY_KEY")
        if key: headers["Authorization"] = f"Bearer {key}"
        r = requests.post(
            f"{os.environ['BW_MODEL_PROXY_URL']}/v1/chat/completions",
            headers=headers,
            json={"model": os.environ["BW_MODEL_MODEL"],
                  "messages": messages, **kwargs},
            timeout=180,
        )
        r.raise_for_status()
        return r.json()["choices"][0]["message"]

    class Task(BenchTask):
        def extract(self, path, index):
            with open(path) as f:
                rows = [json.loads(line) for line in f]
            row = rows[index]
            return {
                "task_id":  row["task_id"],
                "input":    row["input"],
                "expected": row["expected"],
            }

        # ── solve: dispatch on kind ──────────────────────────────
        def solve(self, index, input):
            kind = input["kind"]
            if kind == "multiple_choice": return self._solve_mc(input)
            if kind == "function_call":   return self._solve_fc(input)
            if kind == "math":            return self._solve_math(input)
            if kind == "code":            return self._solve_code(input)
            raise ValueError(f"unknown kind: {kind}")

        def _solve_mc(self, inp):
            prompt = inp["question"] + "\\n\\n"
            for i, c in enumerate(inp["choices"]):
                prompt += f"{chr(65 + i)}. {c}\\n"
            prompt += "\\nAnswer with the single letter."
            msg = _chat([{"role": "user", "content": prompt}])
            m = re.search(r"\\b([A-E])\\b", msg.get("content", ""))
            return m.group(1) if m else ""

        def _solve_fc(self, inp):
            msg = _chat(
                [{"role": "user", "content": inp["question"]}],
                tools=inp["functions"], tool_choice="auto",
            )
            calls = msg.get("tool_calls", [])
            return json.dumps([
                {"name": c["function"]["name"],
                 "args": json.loads(c["function"]["arguments"])}
                for c in calls
            ])

        def _solve_math(self, inp):
            prompt = inp["question"] + "\\n\\nGive only the final numeric answer."
            msg = _chat([{"role": "user", "content": prompt}])
            text = (msg.get("content") or "").strip()
            m = re.search(r"-?\\d+(?:\\.\\d+)?", text.replace(",", ""))
            return m.group(0) if m else text

        def _solve_code(self, inp):
            msg = _chat([{"role": "user", "content": inp["prompt"]}])
            return msg.get("content", "")

        # ── score: dispatch on kind too ──────────────────────────
        def score(self, index, output, expected):
            kind = expected["kind"]
            if kind == "multiple_choice": return self._score_mc(output, expected)
            if kind == "function_call":   return self._score_fc(output, expected)
            if kind == "math":            return self._score_math(output, expected)
            if kind == "code":            return self._score_code(output, expected)
            return 0.0

        def _score_mc(self, output, expected):
            return 1.0 if output.strip().upper() == expected["value"].strip().upper() else 0.0

        def _score_fc(self, output, expected):
            try:
                got  = {(c["name"], json.dumps(c["args"], sort_keys=True))
                        for c in json.loads(output)}
                want = {(c["name"], json.dumps(c["args"], sort_keys=True))
                        for c in expected["value"]}
                return 1.0 if got == want else 0.0
            except Exception:
                return 0.0

        def _score_math(self, output, expected):
            try:
                return 1.0 if abs(float(output) - float(expected["value"])) < 1e-6 else 0.0
            except ValueError:
                return 0.0

        def _score_code(self, output, expected):
            # Strip fenced code blocks if the model wrapped them.
            m = re.search(r"```(?:python)?\\n(.+?)```", output, re.S)
            candidate = m.group(1) if m else output
            with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
                f.write(candidate + "\\n\\n" + expected["test_code"])
                path = f.name
            try:
                subprocess.run(["python3", path], check=True, timeout=30,
                               capture_output=True)
                return 1.0
            except Exception:
                return 0.0
''').strip()

impl = bw.impls.create(
    benchmark=bm.id,
    subset=sub.id,
    label="dispatching scorer v1",
    language="python",
    code=code,
    description="Routes solve + score on task.kind: MC, function-call, math, code.",
)

8.5 Run it and break down per kind

# Pin and launch. Same single run, many domains.
run = bw.runs.launch(
    benchmark=bm.id,
    subset=sub.id,
    impl=impl.id,
    sut={"model": "qwen/qwen3-235b-a22b-thinking", "provider": "openrouter"},
    limits={"run": {"cost": 30.0}},
)
bw.runs.wait(run.id)

# Overall score.
t = bw.runs.tasks(run.id)
print(f"aggregate: {t.summary.pass_rate:.1%}  ({t.summary.passed}/{t.summary.total})")

# Per-kind breakdown. Task rows are slim (no input payload), but the
# namespaced task_id carries the source benchmark, and SOURCES maps
# that to a kind.
from collections import defaultdict
kind_of = {s["bm"]: s["kind"] for s in SOURCES}
by_kind = defaultdict(lambda: [0, 0])  # [passed, total]
for row in t.data:
    kind = kind_of[row.task_id.split("::", 1)[0]]
    by_kind[kind][1] += 1
    if row.score >= 1.0:
        by_kind[kind][0] += 1

print()
for kind, (passed, total) in sorted(by_kind.items()):
    pct = passed / total if total else 0
    print(f"  {kind:<17} {pct:>6.1%}  {passed}/{total}")

Output looks like:

aggregate: 38.2%  (76/199)

  code              42.9%  21/49
  function_call     31.0%  18/58
  math              47.5%  19/40
  multiple_choice   34.6%  18/52

8.6 Publishing and iterating

The composite publishes like any other benchmark: flip the reference run public and the registry derives the rest (§6.5):

bw.runs.update(run.id, visibility="public")

Because the subset is frozen and impl.id is stable, every future run against this gauntlet is directly comparable. The leaderboard at https://benchwright.ai/registry/{bm.id}/{sub.slug} surfaces the aggregate pass rate and the per-kind breakdown together.

When a new generation of models ships, you have two extension paths:

  1. Refresh the gauntlet by re-running §8.2 with an updated panel and publishing v2-<date> as a new subset. The old subset stays addressable so historical comparisons don't move.
  2. Keep the gauntlet, add a harder shell by filtering v1's tasks to ones the new panel still universally fails, and publishing that as a subset slug like v1-core inside the same benchmark.

Either way, the composite benchmark itself is a stable entity that accretes subsets over time. The pattern extends naturally: as soon as enough capability areas have their own public benchmarks, you can build multiple composites targeting different slices (reasoning-only, tool-use-only, hybrid) all from the same source sweeps.

9. Rolling benchmarks

Availability. One gap below: replay.task_ids (scoping a run to just the fresh rows) is a planned replay field the launcher doesn't honor yet, so a replay re-runs the whole subset.

Some benchmarks are never done. A bug-tracker feed, a freshly-scraped CAPTCHA dataset, a weekly code review leaderboard. Create the subset once with rolling=true and append to it forever. A subset created without rolling=true is frozen at creation; appends are rejected with 409 subset_frozen (freezing a rolling subset later is a one-way rolling: false patch).

bm = bw.benchmarks.create(name="Live Bug Repro", category="code",
                           tags=["rolling", "code-review"])
sub = bw.benchmarks.subsets.create(
    benchmark=bm.id, slug="daily",
    name="Daily accretion",
    rolling=True,
)

Ship the scorer once (same shape as §6.3), then run the nightly:

import datetime

def nightly():
    today = datetime.date.today()
    new_rows = fetch_from_bug_tracker(since="yesterday")
    if not new_rows:
        return

    # Append new tasks. Idempotent per day.
    bw.benchmarks.subsets.append_tasks(
        sub.id,
        tasks=[{
            "task_id": r["id"],
            "input":   {"repo": r["repo"], "failing_test": r["test"]},
            "expected": r["fixing_patch_sha"],
        } for r in new_rows],
        idempotency_key=f"bugs-append-{today}",
    )

    # Run the standing panel against yesterday's anchor run.
    for model in ("openai/gpt-5", "anthropic/claude-sonnet-4-6"):
        bw.runs.launch(
            replay={"from_run": anchor_run_id},
            sut={"model": model, "provider": "openrouter"},
            idempotency_key=f"bugs-panel-{today}-{model}",
        )

Once replay.task_ids lands, the nightly will scope each panel run to exactly the fresh rows instead of the whole subset.

10. Cross-model sweeps

Launch a matrix of models against the same benchmark. The anchor run authors the scorer once; every replay reuses it, so the sweep's run ids all share one impl.id.

SWEEP = "bfcl-pfc-qwen3-family-2026-04"
models = ["qwen/qwen3-8b", "qwen/qwen3-14b", "qwen/qwen3-32b",
          "qwen/qwen3-235b-a22b-thinking"]

# 1. Fan out. `replay` anchors all runs to the same impl from a
#    reference run, so costs stay low after the first run authors the
#    scorer.
anchor = bw.runs.launch(
    request="Run BFCL v4 parallel function calling on " + models[0],
    sut={"model": models[0], "provider": "openrouter"},
    idempotency_key=f"sweep-{SWEEP}-{models[0]}",
)
bw.runs.wait(anchor.id)

peers = [anchor.id]
for m in models[1:]:
    r = bw.runs.launch(
        replay={"from_run": anchor.id},
        sut={"model": m, "provider": "openrouter"},
        idempotency_key=f"sweep-{SWEEP}-{m}",
    )
    peers.append(r.id)

# 2. Wait for the sweep, then aggregate. The run ids collected at
#    launch time are the sweep's correlation handle.
rows = []
for rid in peers:
    r = bw.runs.wait(rid)
    t = bw.runs.tasks(rid).summary
    rows.append((r.model.identifier, t.pass_rate, t.passed, t.total,
                 r.cost.billed.amount_cents))
rows.sort(key=lambda x: -x[1])

print("BFCL v4 parallel function calling, Qwen3 family")
for model, pr, p, n, cents in rows:
    print(f"  {model:<44} {pr:>6.1%}  {p}/{n}  ${cents/100:.2f}")

Why this works: every run in the sweep shares impl.id (the anchor's impl), so the registry's same-impl comparison view "just works" for the whole sweep. If you used natural-language launches instead, every run would get its own impl.id from a fresh Atomize phase and the comparison would be muddy.

11. CI gates

For gating a PR on a benchmark score, run the CLI inside the CI job: --gate makes the process exit non-zero when the run's pass rate lands below the threshold, so the CI step itself is the check.

11.1 Gate synchronously with the CLI (works today)

# .github/workflows/benchwright.yml (excerpt)
- run: |
    benchwright run \
      --replay-from "$BW_ANCHOR_RUN" \
      --model qwen/qwen3-32b@openrouter \
      --limit-to run.cost=15 \
      --gate pass_rate=0.90 \
      --idempotency-key "gha-$GITHUB_RUN_ID"

The run streams to the job log and the step fails if the gate doesn't hold (or the run errors). $BW_ANCHOR_RUN is the reference run you're regression-testing against (§10's anchor pattern); --idempotency-key keyed to the CI job id means a re-run of the workflow reattaches to the same Benchwright run instead of paying for a second one.

11.2 Webhooks (preview, not yet available)

Not yet available. The /v1/webhooks API (and the CLI hooks group) is designed but not live yet — benchwright hooks … says so and the endpoints 404. The section below documents the intended contract so integrations can be sketched against it; use §11.1's synchronous gate in the meantime.

For hands-free gating, register a webhook once and let it pass/fail the check when the run settles:

wh = bw.webhooks.create(
    url="https://ci.my-org.com/hooks/benchwright",
    events=["run.finished", "run.failed"],
    description="PR gate",
)
print("signing secret:", wh.signing_secret)   # store in CI env as BW_WEBHOOK_SECRET

Launch with --no-wait (returns immediately after boot) and set the CI check to pending; the webhook handler flips it:

import hmac, hashlib, json, os
from fastapi import FastAPI, Request, Header, HTTPException

app = FastAPI()
SECRET = os.environ["BW_WEBHOOK_SECRET"]

def verify(body: bytes, header: str) -> bool:
    parts = dict(kv.split("=", 1) for kv in header.split(","))
    expected = hmac.new(
        SECRET.encode(), f"{parts['t']}.{body.decode()}".encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(expected, parts["v1"])

@app.post("/hooks/benchwright")
async def hook(req: Request, x_benchwright_signature: str = Header(...)):
    body = await req.body()
    if not verify(body, x_benchwright_signature):
        raise HTTPException(400, "bad signature")

    evt = json.loads(body)
    run = evt["data"]
    md = run.get("metadata", {})
    if "pr" not in md:
        return {"ok": True}

    pr = md["pr"]
    sha = md["commit_sha"]

    if evt["type"] == "run.failed":
        set_github_check(sha, state="failure",
                         description=f"Benchwright run errored: {run['error']}",
                         details_url=run["links"]["dashboard"])
        return {"ok": True}

    tasks = bw.runs.tasks(run["id"]).summary
    gate_ok = tasks.pass_rate >= 0.90
    set_github_check(
        sha,
        state="success" if gate_ok else "failure",
        description=f"{tasks.pass_rate:.1%} pass on BFCL PFC",
        details_url=run["links"]["dashboard"],
    )
    return {"ok": True}

Webhooks will be at-least-once, so every handler must be idempotent (dedupe by X-Benchwright-Delivery-Id). Signing + retry semantics land in the API reference when the endpoints ship.

12. Cost controls

Three levers, from coarse to fine.

12.1 Account-level: balance, cap, auto-refill

bw.billing.controls.update(
    monthly_cap_cents=10_000,
    auto_refill_enabled=True,
    auto_refill_amount_cents=2_500,
    auto_refill_threshold_cents=500,
    alert_enabled=True,
    alert_at_cap_pct=80,
)

Runs that would exceed the monthly cap are halted mid-flight by the live-debit loop. Auto-refill kicks in when balance drops below the threshold; requires a saved payment method.

12.2 Per-run: limits

bw.runs.launch(
    request="...",
    limits={
        "run":     {"cost": 5.0, "duration": 600, "tasks": 50},
        "task":    {"cost": 0.50, "attempts": 1},
        "attempt": {"duration": 120},
    },
)

Dollars and seconds, never cents or milliseconds: the unit belongs to the quantity, so there is nothing in the name to get backwards. task ceilings fail only that task and never retry it; run and phase ceilings stop the run. Full matrix in the CLI reference.

The older constraints={"max_cost_cents": 500, "max_duration_ms": 600_000, "max_tasks": 50} still works and sets the same run-level ceilings — but note those keys carry their units, so 500 there is the same five dollars as 5.0 above.

The pipeline polls the ceilings every 2 seconds and halts when one is breached, with halt_reason of limit_exceeded:<level>.<quantity> — the suffix names which ceiling stopped the run.

A ceiling can also be declared on the ai_model connection itself, under metadata.limits, in the same vocabulary. Only run.model_cost is meaningful there: it bounds the model spend that connection's key pays for, on every run through it, so it is the one to reach for when the thing you need bounded is your provider's bill rather than one launch. A connection may not declare run.cost, which would bound our sandbox and driver spend too — money the credential has nothing to do with. Both a launch ceiling and a connection ceiling may be set; whichever is reached first wins. run.tasks doesn't halt anything: it caps how many samples Materialize runs up front, exactly like metadata.limit (the narrower of the two wins).

A ceiling is where teardown starts, not a figure the run is guaranteed to stay under. The cost check fires at or above the declared run.cost, and halts land between phases, so a single-phase harness run can keep spending until that phase ends and settle a few cents over. Pair a ceiling with max_tasks for a hard bound.

12.3 Spend report

bw.billing.balance.get()
# { 'balance': {'amount_cents': 42155, 'currency': 'USD'},
#   'month_spent': {'amount_cents': 1230, ...},
#   'monthly_cap': {'amount_cents': 10000, ...}, ... }

bw.billing.transactions.list(
    filters={"kind": "trace_debit"},   # one exact kind per query
    limit=200,
)
# per-run debits with reference_id pointing back to the run

Dump everything to CSV for accounting:

with open("bw-2026-04.csv", "wb") as f:
    f.write(bw.billing.export(type="transactions"))

13. Managing provider credentials

The API never sees your provider keys. Secrets live in a vault; a server-side injection proxy attaches them to outbound HTTP from the pipeline, keyed by hostname.

13.1 Add a secret

bw.secrets.create(
    name="OPENROUTER_API_KEY",
    description="personal workspace key, sandboxed org",
    value="sk-or-...",
    rules=[{
        "host_pattern": "openrouter.ai",
        "inject_kind":  "bearer",
        "inject_target": "Authorization",
    }],
)

The rule means: whenever the pipeline makes a request to openrouter.ai, attach Authorization: Bearer sk-or-.... value is write-only; list/read responses carry the secret's name and rules but never the value.

Supported inject_kind values:

kind effect
bearer Authorization: Bearer <value>
header <inject_target>: <value>
basic HTTP basic auth (value formatted username:password)
query ?<inject_target>=<value> on outbound URLs

13.2 Rotate

bw.secrets.update("sec_01J...", value="sk-or-new-key")

Running runs keep using the old value (proxy caches per-run). New runs pick up the rotation.

13.3 Host policy (outbound firewall)

The proxy is default-allow for common hosts (GitHub, PyPI, Hugging Face) and anything with a matching secret rule. Anything else is logged to blocked_hosts for trust-on-observation.

# What did the last run try to reach that we blocked?
print(bw.runs.blocked_hosts(run.id))

# Decide to allow it going forward.
bw.host_policy.add("api.wolframalpha.com", mode="allow")

# Or flip the default-mode to "block" and whitelist deliberately.
bw.host_policy.default("block")
bw.host_policy.starter_pack()   # adds github/pypi/huggingface/etc.

14. Python SDK quick reference

Grouped by resource. Entries marked ✱ are shipped in the SDK but their server endpoint isn't live yet (they 404 today).

idempotency_key= is a parameter of the creation calls, not of every method: runs.launch, runs.fork, benchmarks.create, subsets.create, subsets.append_tasks, impls.create, secrets.create, secrets.update, host_policy.add, billing.controls.update, billing.topups.create and webhooks.create take it. Everything else does not, and the **patch-style updaters (benchmarks.update, subsets.update, impls.update, webhooks.update) will quietly send it as a body field rather than a header. Retry those by re-issuing the patch: they are idempotent by construction.

# Runs
bw.runs.launch(request=..., sut=..., harness=..., compute=..., driver=..., hints=..., constraints=..., trials=..., phase_models=..., metadata=..., idempotency_key=..., timeout=300)
bw.runs.launch(benchmark=..., subset=..., impl=..., sut=...)   # pinned catalog id (+ optional subset/impl)
bw.runs.launch(replay={"from_run": ...}, sut=...)
bw.runs.get(run_id, expand=[...], fields=[...])
bw.runs.list(filters={...}, sort="-started_at", limit=50)      # filters: status, benchmark, visibility, started_at, finished_at
bw.runs.stream(run_id)                           # SSE iterator, live
bw.runs.events(run_id, after_seq=None, types=None, limit=500)  # the same events, paged
bw.runs.wait(run_id, timeout=None, poll_interval=2.0)
bw.runs.fork(run_id, description=..., from_phase=...)
bw.runs.halt(run_id, reason=...)                 # STOP it; the run stays readable (CLI: run kill / run halt)
bw.runs.kill(run_id)                             # DESTROY it, irreversible (CLI: run delete)
bw.runs.update(run_id, visibility=..., review_resolution=..., metadata=..., if_match=...)
bw.runs.publish(run_id, visibility="public")     # the benchmark follows automatically
bw.runs.narrate(run_id, slice="recent")
bw.runs.blocked_hosts(run_id)

# Tasks
tasks = bw.runs.tasks(run_id, passed=None, version=None, limit=100, page_token=None, sort=None)
tasks.summary                                    # aggregate
tasks.data                                       # slim task rows (no payloads)
tasks.payload(tsk_id)                            # full input/expected/output

# Artifacts
bw.runs.artifacts(run_id, type=None, include_versions=False)
bw.runs.artifact(run_id, art_id)
bw.runs.artifact_content(run_id, art_id)         # raw bytes
bw.runs.report(run_id)                           # shortcut to HTML report
bw.runs.fingerprint(run_id)

# Registry (read)
bw.registry.benchmarks.list(tag=None, sort=None, limit=50)
bw.registry.benchmarks.get(bm_id, expand=...)
bw.registry.benchmarks.find(name=...)            # client-side scan of the first 100 rows
bw.registry.subsets.get(sub_id)
bw.registry.subsets.leaderboard(sub_id)
bw.registry.subsets.runs(sub_id)                 # every run against this subset
bw.registry.impls.get(imp_id)
bw.registry.impls.code(imp_id)                   # ✱ task_impl.py text
bw.registry.impls.runs(imp_id)                   # every run scored by this impl

# Authoring
bw.benchmarks.create(name=..., description=..., category=..., tags=...)
bw.benchmarks.update(bm_id, name=..., description=..., category=...)
bw.benchmarks.delete(bm_id)                      # ✱
bw.benchmarks.subsets.create(benchmark=..., slug=..., name=..., tasks=[...], rolling=False)
bw.benchmarks.subsets.update(sub_id, name=..., description=..., rolling=False)
bw.benchmarks.subsets.append_tasks(sub_id, tasks=[...], idempotency_key=...)
bw.benchmarks.subsets.delete(sub_id)             # ✱
bw.impls.create(subset=..., label=..., code=...)
bw.impls.create_from_run(from_run=..., benchmark=..., subset=..., label=...)   # ✱
bw.impls.update(imp_id, label=..., description=...)
bw.impls.delete(imp_id)                          # ✱

# Secrets + host policy
bw.secrets.list()
bw.secrets.get(sec_id)                           # masked preview, never the value
bw.secrets.create(name=..., value=..., rules=[...])
bw.secrets.update(sec_id, value=...)             # rotate
bw.secrets.delete(sec_id)
bw.host_policy.list()
bw.host_policy.add(host_pattern, mode="allow"|"block")
bw.host_policy.delete(entry_id)
bw.host_policy.default("allow"|"block")
bw.host_policy.starter_pack()
bw.host_policy.blocked_hosts()                   # account-wide, vs runs.blocked_hosts(run_id)

# Billing
bw.billing.balance.get()
bw.billing.controls.update(**fields)
bw.billing.topups.create(amount_cents, flow="hosted")
bw.billing.topups.get(top_id)
bw.billing.transactions.list(filters={"kind": ...}, limit=..., page_token=...)
bw.billing.usage_trend()
bw.billing.export(type="transactions"|"runs")

# Webhooks — ✱ whole group (API designed, not live; §11.2)
bw.webhooks.create(url=..., events=[...])
bw.webhooks.list()
bw.webhooks.get(whk_id)
bw.webhooks.update(whk_id, **patch)
bw.webhooks.delete(whk_id)
bw.webhooks.rotate_secret(whk_id)
bw.webhooks.test_delivery(whk_id, event=...)
bw.webhooks.deliveries(whk_id, status=None)
bw.webhooks.retry_delivery(whk_id, whd_id)
bw.webhooks.construct_event(payload, sig_header, secret)   # local HMAC verify, no HTTP

# Settings
bw.settings.list()
bw.settings.get(key)
bw.settings.set(key, value)
bw.settings.delete(key)
bw.settings.schema()                             # the server-side allowlist

# Account / introspection
bw.me()                                          # ✱ use billing.balance.get() today

Three shapes worth knowing before you call them:

15. CLI quick reference

The CLI is a single statically-linked Go binary. Install once, run benchwright login, then any of:

benchwright run "<request>"                                # natural-language launch
benchwright run "<request>" --limit-to run.tasks=5                      # cap samples (metadata.limit)
benchwright run --replay-from <uuid> --model anthropic/claude-haiku-4-5@openrouter
benchwright run launch "Run MMLU on qwen3-32b via Inspect" --harness inspect --harness-task inspect_evals/mmlu_0_shot
benchwright run --benchmark bm_... --subset sub_... --impl latest:official --model qwen/qwen3-32b@openrouter   # pinned
benchwright run show <uuid>
benchwright run tasks <uuid> --passed=false --limit 10
benchwright run fork <uuid> -m "tighter scoring"
benchwright run publish <uuid>
benchwright run kill <uuid>                    # stop it — the run stays readable
benchwright run delete <uuid>                  # DESTROY it — irreversible
benchwright run ls --status running

# bench authoring
benchwright bench create "BFCL v4 mirror" --category tool-use --tag function-calling
benchwright bench subset create bm_<uuid> --slug parallel-function-v4 --file tasks.jsonl --grader-file grader.json
benchwright bench subset append <subset_id> --file more-tasks.jsonl
benchwright bench impl create <subset_id> --file task_impl.py --label "strict-set scorer v1"
benchwright bench impl ls <subset_id>
benchwright bench range <bm_id> 0-5                        # show runs covering an ad-hoc task range
# publishing is run-derived: `benchwright run publish <run_id>`, not a benchmark patch (§6.5)

benchwright secrets add OPENROUTER_API_KEY sk-or-… --rule openrouter.ai:bearer:Authorization
benchwright secrets rotate OPENROUTER_API_KEY sk-or-new
benchwright secrets ls

benchwright billing balance
benchwright billing topup 50
benchwright billing export --type transactions > ledger.csv

benchwright hooks add https://... --events run.finished,run.failed   # not yet available (§11.2)
benchwright hooks test whk_01J… --event run.finished                 # not yet available (§11.2)

@last resolves to the run most recently launched or forked by this CLI (a per-user cache shared across shells), so benchwright run fork @last -m "..." works without copy-pasting an id.

Global flags:

--profile <name>          # select from ~/.config/benchwright/config.toml
--output table|json|ndjson|yaml   # default table on a TTY, ndjson otherwise
--debug                   # include the full problem+json body on errors

Launch-command flags (on run / run launch; run fork has --no-wait too):

--gate pass_rate=0.90     # exit non-zero if summary.pass_rate < 0.90
--no-wait                 # don't stream; print run id and exit
--limit N                 # cap samples (metadata.limit)
--harness / --harness-task

Streaming the run to your terminal is the default for natural-language, pinned, and replay launches; pass --no-wait to opt out.

16. Patterns and gotchas

Prefer replays over natural-language once an anchor run exists. Natural-language launches are 3-10x more expensive because they pay for LLM-driven Find, Analyze, and Atomize. Once one run has authored the scorer, every subsequent run should be replay={"from_run": anchor_id} (or a catalog benchmark pin when you only need the entry, not a prior impl's bytes).

Use idempotency_key everywhere that costs money. Launches, top-ups, subset appends. Retries on a flaky network should never produce two sandboxes or two charges.

Anchor comparisons to a run, not to "the latest impl". Pipeline-authored impls evolve invocation by invocation. If you want apples-to-apples comparisons, keep the anchor run's id and launch every comparison as replay={"from_run": anchor_id} — the replay inherits the anchor's exact impl. (A first-class freeze, impls.create_from_run, is on the roadmap.)

Run ids are your correlation channel. Keep the ids your integration launches (a manifest file, a CI artifact, a database column upstream) and read runs back by id. Launch metadata is behavioral configuration (limit, harness config, autonomous) — it is not persisted as a queryable tag today, so filters={"metadata.sweep": ...} matches nothing. The supported list filters are status, benchmark, visibility, started_at, finished_at.

Webhooks (when they ship) are at-least-once. Handlers must dedupe on X-Benchwright-Delivery-Id. Pair with a short (24h) cache of seen ids in Redis / SQLite / filesystem.

The model is not the Driver. sut (the API's model field) is what the pipeline tests. driver is what the pipeline uses to drive the evaluation (Find/Atomize/Report). Confusing the two is the most common misconfiguration in CI logs. This is also why driver replaced the older agent spelling, and why agent was then removed rather than left as an alias: "agent" now means the Operator, which is neither of these two.

Reach for a third-party harness before authoring a scorer. If the benchmark you want already ships in Inspect, lm-eval-harness, lighteval, or terminal-bench, launch with harness=<harness> and the harness's task id rather than letting Atomize write a scorer from scratch — it's faster, cheaper, and matches the harness's canonical scoring. Reserve benchwright (the default) for benchmarks no harness covers.

Compute is per-run. compute overrides the backend for a single run; it doesn't change your account default. Bring-your-own compute (sandbox0 on your own box) bills $0, exactly like a bring-your-own model key.

A replay compares against the parent's impl, not today's state. If you fix your scorer and want to reassess old runs, fork (which re-authors from your description) and treat the fork as a new anchor; replays of the parent keep measuring with the old scorer — useful when the old scorer is exactly what you want to hold constant.

Visibility is run-derived. Publishing a run (run publish / runs.update(visibility="public")) is what makes its benchmark visible in the registry; benchmarks and subsets don't take a direct visibility patch. Hiding every public run hides the benchmark again.

Delete is rare. Runs are immutable history and the registry has to preserve them to keep leaderboards honest. Hide via visibility, don't delete. The only things you regularly delete are secrets and host-policy entries.

Prefer the first launch and the retry to be the same command. An identical benchwright run invocation re-attaches to the run it already launched (idempotency key derived from argv + cwd + commit) instead of paying for a second one — that's why CI examples key the launch to the CI job id.

Versioning and deprecation

This guide evolves with the API. The API reference documents the deprecation policy and the Sunset header signals that tell clients when an endpoint will go away.