Build a repeated-trial benchmark that records failures and incomplete runs

Problem

When you need this. Use this when sampled model outputs vary across runs and you need repeated trials, resumable trial logs and pass@k versus pass^k metrics.

The versioned evaluation harness established fixed cases, graders and saved model replies. Replaying a reply repeats the same check; a fresh model call can produce a different answer. The same message can be triaged correctly on one call and wrongly on the next, so one trial measures one draw rather than the range of behavior.

Running repeated live trials raises three engineering problems:

• Losing work. The bundled generic harness’s record() keeps every trial in memory and writes nothing until the last one finishes. A crash, a closed laptop or Ctrl-C after an hour of model calls loses the hour.
• Losing trials silently. A run stopped early has fewer trials for some cases. If those trials vanish from the record, the pass rate is computed over a smaller, uneven sample, and nobody can tell.
• Reading the counts. Ten trials per case give a pass count per case. A single pass rate hides whether every case passes most of the time or some cases always fail.

We will build bench.py and report.py beside the harness. The contract:

• Input: the bundled case set and task module, a system that answers one trial, a trial count k and a log file.
• Output: an append-only log with one line per trial, a harness recording made from it, and a report with per-case pass counts, flaky cases, pass@k and pass^k for every k up to the trial count, failures by check and incomplete trials by cause.
• Rules: every planned trial ends as pass, fail or incomplete. A trial that never ran is incomplete, never missing. A rerun with the same log resumes and never repeats a logged trial.

The systems are two local models from Choose a model and runtime for a specific task, gemma4:e2b and qwen3.5:0.8b, triaging the 9 development cases at temperature 1, with 10 trials each. 7 of them call the model; the other 2 are rejections checked before any call. The 7 held-out cases were not run: this build measures how to run and read repeated trials, not either model, so the held-out split stays unused.

Concepts

The repeated trials lesson defined trials, pass@k and pass^k. This build adds the parts needed to measure them from recordings:

Seeded sampling

At temperature 1 the model samples each token. A fixed seed controls the sampling random-number sequence; exact output can still depend on runtime behavior. Here trial t uses seed 1000 + t for every case, so trial 3 of every case uses the same seed, and the trials of one case differ from each other.

Append-only log

A file in JSON Lines format: a header line naming the plan (case-set hash, system, splits, trial count), then one line per trial, written and flushed as soon as the trial ends.

Trial-major order

Trial 0 of every case, then trial 1 of every case, and so on. A run that stops early has run every case the same number of times, plus or minus one.

NotRun

The error type given to a planned trial that the log does not contain. The harness grades it incomplete, so a stopped run cannot pass for a finished one.

Flaky case

A case where some trials that ran passed and some did not.

pass@k and pass^k from n trials

With n recorded trials per case and c passes, pass@k is the chance that at least one of k trials drawn from those n passes: 1 − C(n − c, k) / C(n, k). pass^k is the chance that all k pass: C(c, k) / C(n, k). Both are averaged over cases. C(a, b) counts the ways to choose b of a trials.

Follow one case. password-changed expects account, and its failure criterion fails any output with needs_human set to false. Over 10 trials, qwen3.5:0.8b passes it 7 times. Its other trials returned:

A one-trial evaluation would have reported whichever of these it happened to draw. From the counts, the chance that a single trial passes is 7/10, the chance that at least one of three passes (pass@3) is 0.9917, and the chance that all three pass (pass^3) is 0.2917.

The estimate draws without replacement from the recorded trials. The tempting shortcut, 1 − (1 − c/n)^k, treats the trials as an endless supply; Chen and colleagues show it is biased for pass@k. A test in this build checks both formulas against a count over every subset of trials.

On this machine, both logs were recorded 3 times with the same seeds. outputs_digest.py hashes every trial’s output; the digests in recorded/recordings.json are identical across recordings for both models. These recordings demonstrate matching outputs for this environment and seed list; a seed alone does not guarantee identical output or correctness. It does not make the pass counts a property of the model in general: other seeds, prompts or Ollama versions give other draws.

Architecture

The Case set enters bench.py · run_bench(). next (case, trial) takes the next planned pair in trial-major order and skips any trial the log already holds. budget spent? checks the time budget between trials. When it is spent, the run reaches stop starting: it starts no new trial, and every trial not yet run becomes NotRun. When it is not spent, harness.call() sends the message to the Ollama model with the trial’s seed and receives the JSON reply. The trial’s output or error is appended as one line to the Log, and the cobalt next trial path returns to next (case, trial). A log written for another case set, system, split or trial count is Refused: BenchError.

After the run, to_recording() turns the log into a Recording in the harness’s format, where every planned trial missing from the log is a NotRun error. harness.py run replays it and grades every trial with the task’s checks, unchanged, and report.py writes runs/report.json. The recorded grids show one cell per trial: the stopped gemma4:e2b log, with its NotRun cells, and the finished qwen3.5:0.8b run, with its failed trials.

Working code

Runtime: Python 3.11 or later. Dependencies: Python standard library only. The tests and the report replay the logs; they call no model and no network. To record again you need Ollama with both models at the digests pinned in model_triage.py. The README lists every file.

The system

model_triage.py sends one message per trial to a local model, asks for JSON that matches the task’s output fields with a schema in Ollama’s format field, and returns an Answer with the seed, temperature, token count, stop reason and Ollama version as metadata. Blank or missing input raises ValueError before any model call, exactly as the keyword rules did, so the two rejection cases stay deterministic. A reply cut off at the token limit raises TruncatedReply, and a reply that is not JSON raises json.JSONDecodeError; the harness grades both as incomplete.

"""Inbox triage by a local model, sampled: the system the benchmark repeats.

Each trial sends one message to a local Ollama model at temperature 1 with a
seed that depends only on the trial number, and asks for JSON that matches the
triage task's output fields. Input is checked before any model call, so blank
or missing text raises the same rejections as the keyword rules in the harness
build. A reply cut off at the token limit raises TruncatedReply, and a reply
that is not JSON raises json.JSONDecodeError: both become incomplete trials.
Python standard library only.
"""
import hashlib
import json
import urllib.request

from harness import Answer
from triage_task import CATEGORIES

OLLAMA = "http://127.0.0.1:11434"
# Tag -> first 12 hex digits of the digest the recordings were made with.
MODELS = {"gemma4:e2b": "b37049369adf", "qwen3.5:0.8b": "de63045f2975"}
TEMPERATURE = 1.0
NUM_PREDICT = 64
SEED_BASE = 1000  # Trial t uses seed 1000 + t, the same for every case and model.
PROMPT = (
    "Classify one customer message for an online shop's support inbox. Reply with JSON only.\n"
    "category: billing (charges, refunds, payments), delivery (orders on their way, tracking, arrival), "
    "account (signing in, passwords, account access), or other (anything else).\n"
    "needs_human: true when the message suggests that someone else used or changed the customer's account; "
    "otherwise false.")
SCHEMA = {"type": "object", "required": ["category", "needs_human"],
          "properties": {"category": {"type": "string", "enum": list(CATEGORIES)},
                         "needs_human": {"type": "boolean"}}}


class TruncatedReply(RuntimeError):
    """The model stopped at the token limit, so its reply is incomplete."""


def post(path, body, timeout=120):
    request = urllib.request.Request(f"{OLLAMA}{path}", data=json.dumps(body).encode("utf-8"),
                                     headers={"Content-Type": "application/json"})
    with urllib.request.urlopen(request, timeout=timeout) as response:
        return json.loads(response.read())


def get(path):
    with urllib.request.urlopen(f"{OLLAMA}{path}", timeout=10) as response:
        return json.loads(response.read())


class ModelTriage:
    """A harness system: name, version, mode, model, prompt and tools, plus respond(case, trial)."""

    mode = "live"
    tools = None

    def __init__(self, tag):
        self.tag, self.name, self.version = tag, f"model-triage-{tag.split(':')[0]}", "1"
        self.model = f"{tag}@{MODELS[tag]}"
        self.prompt = f"triage-v1 sha256:{hashlib.sha256(PROMPT.encode()).hexdigest()[:16]}"
        self.server = None

    def check_model(self):
        """Refuse to run if the installed weights differ from the pinned digest."""
        digests = {m["name"]: m["digest"] for m in get("/api/tags")["models"]}
        if not digests.get(self.tag, "").startswith(MODELS[self.tag]):
            raise RuntimeError(f"{self.tag} is not installed with digest {MODELS[self.tag]}")
        self.server = get("/api/version")["version"]

    def seed_for(self, trial):
        return SEED_BASE + trial

    def chat(self, message, seed):
        return post("/api/chat", {
            "model": self.tag, "stream": False, "think": False, "format": SCHEMA,
            "options": {"temperature": TEMPERATURE, "seed": seed, "num_predict": NUM_PREDICT},
            "messages": [{"role": "system", "content": PROMPT}, {"role": "user", "content": message}]})

    def respond(self, case, trial):
        message = case["input"]
        if not isinstance(message, str):
            raise ValueError("message is not text")
        if not message.strip():
            raise ValueError("message is blank")
        if self.server is None:
            self.check_model()
        seed = self.seed_for(trial)
        result = self.chat(message, seed)
        metadata = {"seed": seed, "temperature": TEMPERATURE, "ollama": self.server,
                    "eval_count": result.get("eval_count"), "done_reason": result.get("done_reason")}
        if result.get("done_reason") == "length":
            raise TruncatedReply(f"reply stopped at {NUM_PREDICT} tokens")
        return Answer(json.loads(result["message"]["content"]), metadata)


gemma = ModelTriage("gemma4:e2b")
qwen = ModelTriage("qwen3.5:0.8b")

The benchmark

"""Repeated-trial benchmark: run every case k times, log each trial, resume, stop on budget.

The harness's record() keeps every trial in memory and writes nothing until the
last one finishes, so a crash or a stopped run loses everything. bench.py
appends one JSON line per trial to a log as soon as the trial ends. Running it
again with the same log skips the trials already logged. A time budget stops
the run between trials. to_recording() turns a log into a harness recording in
which every trial that never ran is an error of type NotRun, so the harness
grades it incomplete instead of dropping it.
Python standard library only.
Run: python3 bench.py --system model_triage:gemma --trials 10 --log recorded/gemma.jsonl
"""
import argparse
import json
import platform
import time
from datetime import datetime, timezone
from pathlib import Path

import harness

BENCH_VERSION = "1"
HEADER_KEYS = ("bench_version", "harness_version", "case_set_sha256", "system", "splits", "trials")


class BenchError(ValueError):
    """The log cannot be resumed or converted safely."""


def header_for(system, case_set, splits, trials):
    return {"bench_version": BENCH_VERSION, "harness_version": harness.HARNESS_VERSION,
            "case_set_sha256": case_set["sha256"], "system": harness.system_identity(system),
            "splits": sorted(set(splits)), "trials": trials}


def schedule(case_set, splits, trials):
    """Trial-major order: trial 0 of every case, then trial 1 of every case, and so on.
    A run stopped early then has the same number of trials for every case, plus or minus one."""
    cases = harness.plan(case_set, splits, trials, include_held_out=False)
    return [(case, trial) for trial in range(trials) for case in cases]


def read_log(path):
    """Return (header, rows). A truncated last line, from a crash mid-write, is ignored."""
    lines = Path(path).read_text(encoding="utf-8").splitlines()
    if not lines:
        raise BenchError(f"{path} is empty")
    header, rows = json.loads(lines[0]), []
    for number, line in enumerate(lines[1:], start=2):
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError:
            if number != len(lines):
                raise BenchError(f"{path}: line {number} is not JSON") from None
    return header, rows


def run_bench(system, case_set, splits, trials, log_path, budget_s=None, clock=time.monotonic, hardware=None):
    """Run every planned trial not yet in the log. Return counts of what this call did."""
    log_path = Path(log_path)
    header = header_for(system, case_set, splits, trials)
    done = set()
    if log_path.exists() and log_path.stat().st_size:
        logged, rows = read_log(log_path)
        differs = [key for key in HEADER_KEYS if logged.get(key) != header[key]]
        if differs:
            raise BenchError(f"{log_path} was written for another {', '.join(differs)}; use a new log")
        done = {(row["case_id"], row["trial"]) for row in rows}
        with log_path.open("r+b") as handle:
            data = handle.read()
            if not data.endswith(b"\n"):
                tail_start = data.rfind(b"\n") + 1
                try:
                    json.loads(data[tail_start:])
                except (ValueError, UnicodeError):
                    handle.truncate(tail_start)  # Discard only an incomplete final JSON value.
                else:
                    handle.write(b"\n")  # Preserve a complete row already accepted by read_log().
    else:
        # started_at, budget_s and environment describe this run; a resume does not compare them.
        environment = {"python": platform.python_version(), "platform": platform.platform(), "hardware": hardware}
        log_path.write_text(json.dumps(dict(header, started_at=datetime.now(timezone.utc).isoformat(timespec="seconds"),
                                            budget_s=budget_s, environment=environment), ensure_ascii=False) + "\n",
                            encoding="utf-8")
    start, ran, stopped = clock(), 0, False
    with log_path.open("a", encoding="utf-8") as log:
        for case, trial in schedule(case_set, splits, trials):
            if (case["id"], trial) in done:
                continue
            if budget_s is not None and clock() - start >= budget_s:
                stopped = True
                break
            row = {"case_id": case["id"], "trial": trial}
            call_start = time.perf_counter()
            try:
                output, metadata, latency = harness.call(system, case, trial)
                row.update(output=output, metadata=metadata, latency_s=latency)
            except Exception as error:  # Recorded, not raised: an error is a result.
                row.update(error={"types": harness.error_types(error), "message": str(error)},
                           latency_s=round(time.perf_counter() - call_start, 6))
                if hasattr(system, "seed_for"):  # A failed trial keeps its seed, so it can be rerun alone.
                    row["metadata"] = {"seed": system.seed_for(trial)}
            log.write(json.dumps(row, ensure_ascii=False) + "\n")
            log.flush()  # The trial survives a crash of the next one.
            ran += 1
    planned = len(schedule(case_set, splits, trials))
    return {"planned": planned, "already_logged": len(done), "ran": ran,
            "not_run": planned - len(done) - ran, "stopped_by_budget": stopped}


def to_recording(log_path, case_set):
    """Convert a log into a harness recording. Every planned trial missing from the log
    becomes an error of type NotRun, so the harness counts it as incomplete."""
    header, rows = read_log(log_path)
    if header.get("case_set_sha256") != case_set["sha256"]:
        raise BenchError(f"{log_path} was written for a different case set version")
    by_key = {}
    for row in rows:
        key = (row["case_id"], row["trial"])
        if key in by_key:
            raise BenchError(f"{log_path}: {key[0]} trial {key[1]} is logged twice")
        by_key[key] = {k: row[k] for k in ("output", "metadata", "error", "latency_s") if k in row}
    responses = {}
    for case in harness.plan(case_set, header["splits"], header["trials"], include_held_out=False):
        responses[case["id"]] = [
            by_key.get((case["id"], trial),
                       {"error": {"types": ["NotRun"], "message": "not run: the benchmark stopped first"},
                        "latency_s": 0.0})
            for trial in range(header["trials"])]
    return {"harness_version": header["harness_version"], "system": header["system"],
            "case_set_sha256": header["case_set_sha256"], "splits": header["splits"],
            "trials": header["trials"], "responses": responses}


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("--task", default="triage_task")
    parser.add_argument("--cases", default="cases.json")
    parser.add_argument("--system", help="<module>:<attribute>; omit with --convert")
    parser.add_argument("--trials", type=int, default=10)
    parser.add_argument("--log", required=True)
    parser.add_argument("--budget-s", type=float, help="stop starting new trials after this many seconds")
    parser.add_argument("--hardware", help="a description of the machine, stored in a new log's header")
    parser.add_argument("--convert", metavar="RECORDING", help="write the log as a harness recording and exit")
    args = parser.parse_args(argv)
    case_set = harness.load_cases(args.cases, harness.load_task(args.task))
    if args.convert:
        harness.write(args.convert, to_recording(args.log, case_set))
        print(f"wrote {args.convert}")
        return
    system = harness.load_system(args.system, case_set)
    if hasattr(system, "check_model"):  # Check the digest once, before any trial is logged.
        try:
            system.check_model()
        except RuntimeError as error:
            raise BenchError(str(error)) from None
    counts = run_bench(system, case_set, ["development"], args.trials, args.log, args.budget_s, hardware=args.hardware)
    print(f"{system.name}: planned {counts['planned']}, already logged {counts['already_logged']}, "
          f"ran {counts['ran']}, not run {counts['not_run']}"
          + (" (stopped by the time budget)" if counts["stopped_by_budget"] else ""))


if __name__ == "__main__":
    try:
        main()
    except (harness.CaseError, harness.RunError, BenchError) as error:
        raise SystemExit(f"error: {error}")

Read the important decisions in order:

• run_bench() compares the log’s header with the current plan and refuses a log written for another case set, system, split, trial count or harness version. Resuming into the wrong log would mix two benchmarks.
• On resume it truncates a partial last line, left by a crash in the middle of a write, before appending.
• Each trial is written and flushed before the next one starts. An error from the system is logged as a result; only an interrupt such as Ctrl-C stops the loop, and every finished trial is already on disk.
• The budget is checked between trials, never during one, so a trial is never cut off and logged half-done.
• to_recording() refuses a log with a duplicated trial, and fills every missing trial with a NotRun error.

The report

"""Read repeated-trial run records and report what a single trial would hide.

For each system: verdict counts, failures by check and incomplete trials by
error type, every case's pass count, the flaky cases (some trials pass, some
do not), estimates of pass@k and pass^k for every k up to the trial count, and
the median and 90th-percentile latency of the trials that returned an output.
Trials the benchmark never ran (NotRun) are counted on their own. They are not
flakiness, and a run with any of them gets no pass@k or pass^k: resume it first.
Python standard library only.
Run: python3 report.py runs/gemma.json runs/qwen.json --out runs/report.json
"""
import argparse
import math
from collections import Counter

from harness import RunError, read_json, write

REPORT_VERSION = "1"


def pass_at_k(n, c, k):
    """Chance that at least one of k trials drawn without replacement from n passes (c passed)."""
    return 1.0 - math.comb(n - c, k) / math.comb(n, k)


def pass_power_k(n, c, k):
    """Chance that all k trials drawn without replacement from n pass (c passed)."""
    return math.comb(c, k) / math.comb(n, k)


def nearest_rank(values, share):
    """The smallest value with at least `share` of the values at or below it."""
    ordered = sorted(values)
    return ordered[max(1, math.ceil(share * len(ordered))) - 1]


def first_word(reason):
    return reason.split(":", 1)[0]


def system_report(record):
    manifest, results = record["manifest"], record["results"]
    if manifest["splits"] != ["development"]:
        raise RunError("the benchmark reports development runs only")
    n = manifest["trials"]
    per_case = {}
    for row in results:
        counts = per_case.setdefault(row["case_id"], {"class": row["class"], "pass": 0, "fail": 0, "incomplete": 0,
                                                       "not_run": 0})
        counts[row["verdict"]] += 1
        counts["not_run"] += row["verdict"] == "incomplete" and first_word(row["reason"]) == "NotRun"
    if any(sum(c[v] for v in ("pass", "fail", "incomplete")) != n for c in per_case.values()):
        raise RunError("every case must have exactly one row per trial")
    latencies = [row["metadata"]["recorded_latency_s"] for row in results
                 if row["output"] is not None and "recorded_latency_s" in row["metadata"]]
    not_run = sum(c["not_run"] for c in per_case.values())
    curve = None if not_run else {str(k): {"pass_at_k": round(math.fsum(pass_at_k(n, c["pass"], k) for c in per_case.values()) / len(per_case), 4),
                      "pass_power_k": round(math.fsum(pass_power_k(n, c["pass"], k) for c in per_case.values()) / len(per_case), 4)}
             for k in range(1, n + 1)}
    return {
        "system": manifest["system"],
        "recording_sha256": manifest["recording_sha256"],
        "trials_per_case": n,
        "cases": len(per_case),
        "verdicts": dict(Counter(row["verdict"] for row in results)),
        "not_run": not_run,
        "failures_by_check": dict(sorted(Counter(row["check"] for row in results if row["verdict"] == "fail").items())),
        "incomplete_by_error": dict(sorted(Counter(first_word(row["reason"]) for row in results
                                                   if row["verdict"] == "incomplete").items())),
        "per_case": per_case,
        # Flaky: among the trials that ran, some passed and some did not.
        "flaky_cases": [case for case, c in per_case.items() if 0 < c["pass"] < n - c["not_run"]],
        "never_passed": [case for case, c in per_case.items() if c["pass"] == 0],
        "curve": curve,
        "latency_s": ({"trials": len(latencies), "median": round(nearest_rank(latencies, 0.5), 3),
                       "p90": round(nearest_rank(latencies, 0.9), 3)} if latencies else None),
    }


def build_report(records):
    task = {r["manifest"]["task"]["sha256"] for r in records}
    cases = {r["manifest"]["case_set"]["sha256"] for r in records}
    if len(task) != 1 or len(cases) != 1:
        raise RunError("all run records must use the same task module and case set")
    return {"report_version": REPORT_VERSION, "case_set": records[0]["manifest"]["case_set"],
            "systems": [system_report(record) for record in records]}


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument("runs", nargs="+")
    parser.add_argument("--out", required=True)
    args = parser.parse_args(argv)
    report = build_report([read_json(path) for path in args.runs])
    write(args.out, report)
    for system in report["systems"]:
        last = str(system["trials_per_case"])
        estimate = (f"pass@{last} {system['curve'][last]['pass_at_k']}, pass^{last} {system['curve'][last]['pass_power_k']}"
                    if system["curve"] else f"no pass@k: {system['not_run']} trials not run")
        print(f"{system['system']['name']}: {system['verdicts']}; flaky {len(system['flaky_cases'])} of "
              f"{system['cases']} cases; {estimate}")


if __name__ == "__main__":
    try:
        main()
    except RunError as error:
        raise SystemExit(f"error: {error}")

report.py counts NotRun trials apart from other incomplete trials. A case with unrun trials is not flaky, and a run with any unrun trial gets no pass@k or pass^k: the estimates need the same n for every case.

Record, stop and resume

The recorded run used a 30-second budget for the first gemma4:e2b run, stored in the log’s header as budget_s, then resumed the same log:

python3 bench.py --system model_triage:gemma --trials 10 --log recorded/gemma.jsonl --budget-s 30
python3 bench.py --system model_triage:gemma --trials 10 --log recorded/gemma.jsonl
python3 bench.py --system model_triage:qwen --trials 10 --log recorded/qwen.jsonl

When the budget stopped the first run, the log held 59 trials and 31 were not run. Trial-major order spread them evenly: every case had run 6 to 7 times. Converted and graded at that point, the stopped run reports 31 incomplete trials with the cause NotRun, 0 flaky cases and no pass@k. The second command skipped the logged trials and ran the rest. The README lists the commands that convert the logs, run the harness and write the reports.

Read the report

gemma4:e2b passed 90 of 90 trials. qwen3.5:0.8b is flaky on 3 cases, and product-advice passed only 2 of 10 times. Its pass@10 is 1, the share of cases that passed in at least one trial, while its pass^10 is 0.6667, the share that passed every trial.

A single-trial evaluation of qwen3.5:0.8b would have reported a different count for each seed. Across its 10 trials, the number of development cases that passed in that trial ranged from 6 to 9 of 9.

The finished runs of gemma4:e2b and qwen3.5:0.8b have 0 and 0 incomplete trials: no reply was cut off and every reply was JSON. The incomplete trials in this build are the NotRun trials of the stopped run. The latencies are the time of each model call with Ollama 0.32.14 on one machine (your numbers will differ), recorded after waiting until no other program was using the Ollama server. They cover the 70 trials per model that called the model; the two rejection cases never do.

{
  "report_version": "1",
  "case_set": {
    "name": "inbox-triage",
    "sha256": "9b7392a423cd6595973591357ab12843404f9d8d69ef3b5b697aca6bc4bf1cf6"
  },
  "systems": [
    {
      "system": {
        "name": "model-triage-gemma4",
        "version": "1",
        "mode": "replay",
        "model": "gemma4:e2b@b37049369adf",
        "prompt": "triage-v1 sha256:053fdf51b21d891e",
        "tools": null
      },
      "recording_sha256": "5091cc6f4507bb6178eb49e30166904c48ff382d1e20b6894b1aabdea7557fb6",
      "trials_per_case": 10,
      "cases": 9,
      "verdicts": {
        "pass": 90
      },
      "not_run": 0,
      "failures_by_check": {},
      "incomplete_by_error": {},
      "per_case": {
        "charged-twice": {
          "class": "billing",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "stuck-tracking": {
          "class": "delivery",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "password-changed": {
          "class": "account security",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "product-advice": {
          "class": "out of scope",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "refund-missing": {
          "class": "billing",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "charged-after-closing": {
          "class": "billing, ambiguous wording",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "blank-message": {
          "class": "empty input",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "missing-body": {
          "class": "missing input",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "late-order": {
          "class": "delivery",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        }
      },
      "flaky_cases": [],
      "never_passed": [],
      "curve": {
        "1": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "2": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "3": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "4": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "5": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "6": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "7": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "8": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "9": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        },
        "10": {
          "pass_at_k": 1.0,
          "pass_power_k": 1.0
        }
      },
      "latency_s": {
        "trials": 70,
        "median": 0.651,
        "p90": 0.719
      }
    },
    {
      "system": {
        "name": "model-triage-qwen3.5",
        "version": "1",
        "mode": "replay",
        "model": "qwen3.5:0.8b@de63045f2975",
        "prompt": "triage-v1 sha256:053fdf51b21d891e",
        "tools": null
      },
      "recording_sha256": "b314b3dc5d1110821cfed88b21b39bc0212244e9eb5c3e6f51d695fc6fd38d6b",
      "trials_per_case": 10,
      "cases": 9,
      "verdicts": {
        "pass": 77,
        "fail": 13
      },
      "not_run": 0,
      "failures_by_check": {
        "failure-criteria": 2,
        "triage-outcome": 11
      },
      "incomplete_by_error": {},
      "per_case": {
        "charged-twice": {
          "class": "billing",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "stuck-tracking": {
          "class": "delivery",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "password-changed": {
          "class": "account security",
          "pass": 7,
          "fail": 3,
          "incomplete": 0,
          "not_run": 0
        },
        "product-advice": {
          "class": "out of scope",
          "pass": 2,
          "fail": 8,
          "incomplete": 0,
          "not_run": 0
        },
        "refund-missing": {
          "class": "billing",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "charged-after-closing": {
          "class": "billing, ambiguous wording",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "blank-message": {
          "class": "empty input",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "missing-body": {
          "class": "missing input",
          "pass": 10,
          "fail": 0,
          "incomplete": 0,
          "not_run": 0
        },
        "late-order": {
          "class": "delivery",
          "pass": 8,
          "fail": 2,
          "incomplete": 0,
          "not_run": 0
        }
      },
      "flaky_cases": [
        "password-changed",
        "product-advice",
        "late-order"
      ],
      "never_passed": [],
      "curve": {
        "1": {
          "pass_at_k": 0.8556,
          "pass_power_k": 0.8556
        },
        "2": {
          "pass_at_k": 0.921,
          "pass_power_k": 0.7901
        },
        "3": {
          "pass_at_k": 0.9472,
          "pass_power_k": 0.7509
        },
        "4": {
          "pass_at_k": 0.963,
          "pass_power_k": 0.7222
        },
        "5": {
          "pass_at_k": 0.9753,
          "pass_power_k": 0.7006
        },
        "6": {
          "pass_at_k": 0.9852,
          "pass_power_k": 0.6852
        },
        "7": {
          "pass_at_k": 0.9926,
          "pass_power_k": 0.675
        },
        "8": {
          "pass_at_k": 0.9975,
          "pass_power_k": 0.6691
        },
        "9": {
          "pass_at_k": 1.0,
          "pass_power_k": 0.6667
        },
        "10": {
          "pass_at_k": 1.0,
          "pass_power_k": 0.6667
        }
      },
      "latency_s": {
        "trials": 70,
        "median": 0.403,
        "p90": 0.462
      }
    }
  ]
}

Check your understanding

Try it: the report gives late-order 8 passes in 10 trials for qwen3.5:0.8b. Before you compute anything, predict whether its pass^5 is closer to 0.8 or to 0.2. Then compute C(c, 5) / C(n, 5).

Failure cases

Each behavior below has a named test in test_bench.py.

test_recorded_runs_and_report_match_fresh_runs converts all three logs, replays them through the harness, rebuilds both reports and compares everything with the saved files after removing only RUN_DESCRIPTION_FIELDS. test_report_counts_flaky_cases_and_latency checks the flaky rule, the curve and the nearest-rank percentile on a scripted system.

python3 -m unittest -v test_bench.py

The run ends with Ran 15 tests and OK on Python 3.11 and 3.15.

Deployment

The deployment unit is a directory run from the command line. On the machine that will run benchmarks, install Ollama and both models, check the digests with ollama list, then run the tests and one short benchmark with a budget to confirm the server answers:

python3 -m unittest test_bench.py
python3 bench.py --system model_triage:qwen --trials 1 --log /tmp/smoke.jsonl --budget-s 20

bench.py prints one line per invocation with the planned, already logged, run and not-run counts, and exits with status 1 and an error: line when it refuses a log. Keep the logs with the run records and reports: the log is the original trial record, and every other file is rebuilt from it. A long benchmark can be stopped at any time and resumed with the same command.

Model calls on a shared server compete with each other, which changes latency and can cause timeouts that the benchmark records as incomplete. Record benchmarks when the server is otherwise idle, or report the load beside the numbers.

Scaling

No throughput or capacity was measured. The current code favors a complete, resumable record:

To measure, fix the cases and the trial count and time bench.py per model on an idle server. The harness’s per-trial latency_s in each log line is the raw data.

Key takeaways

• A sampled model needs several trials per case; one trial measures one draw.
• Log every trial as it ends, in an order that keeps a stopped run even across cases, and resume from the log.
• A trial that never ran is incomplete, not missing, and a run with unrun trials is not reported as a pass rate.
• Estimate pass@k and pass^k from the counts without replacement; the per-case counts show the flaky cases that a total hides.

Best practices

• Check the log’s plan before resuming, and refuse to mix two benchmarks in one log.
• Keep input checks deterministic and outside the model call, so rejection cases do not depend on sampling.
• Record seeds, sampling settings, model digests and the server version with every trial.

References

Checked October 20, 2026:

• Chen et al., Evaluating Large Language Models Trained on Code (2021) — the unbiased pass@k estimator from n samples with c correct (equation 1), and why 1 − (1 − p)^k is biased.
• Yao et al., τ-bench (2024) — the pass^k metric for reliability over repeated trials.
• Anthropic: Demystifying evals for AI agents — trials, pass@k and pass^k for nondeterministic systems.
• Ollama: Chat API — format with a JSON schema, think, and the seed, temperature and num_predict options.
• Google: Gemma 4 model card and Qwen: Qwen3.5-0.8B model card — both Apache 2.0.
• JSON Lines — one JSON value per line.
• Python: math.comb and unittest.mock.

For the definitions of trials, pass@k and pass^k, start with Run repeated trials and compare quality trade-offs.