Self-reported. Reproducible.

#!/usr/bin/env python3
"""c=64 streaming benchmark, matched to bench_results_overnight_20260605T031811Z.

Shape of record: concurrency 64, 256 requests, max_tokens 128, 32 unique prompts
(scripts/bench.py's 20 VARIED + 12 CODING), warmup first, N runs, median across runs.
Streams every response so TTFT is real. Output tokens = content chunks (same
convention as the published CSVs, where output_tokens == chunks).
"""
from __future__ import annotations
import argparse, asyncio, json, statistics, sys, time
import httpx

def pct(xs, p):
    if not xs: return None
    xs = sorted(xs); k = (len(xs) - 1) * p
    f = int(k); c = min(f + 1, len(xs) - 1)
    return xs[f] + (xs[c] - xs[f]) * (k - f)

async def one(client, url, headers, body, idx):
    t0 = time.perf_counter(); ttft = None; chunks = 0; usage = None; err = None; hdrs = None
    try:
        async with client.stream("POST", url, headers=headers, json=body) as r:
            hdrs = {k.lower(): v for k, v in r.headers.items() if k.lower().startswith(("x-", "server"))}
            if r.status_code != 200:
                err = f"http{r.status_code}:{(await r.aread())[:120].decode('utf-8','replace')}"
            else:
                async for line in r.aiter_lines():
                    if not line.startswith("data:"): continue
                    data = line[5:].strip()
                    if data == "[DONE]": break
                    try: obj = json.loads(data)
                    except Exception: continue
                    if obj.get("usage"): usage = obj["usage"]
                    for ch in obj.get("choices") or []:
                        d = ch.get("delta") or {}
                        piece = d.get("content") or d.get("reasoning_content") or d.get("reasoning") or ""
                        if piece:
                            if ttft is None: ttft = (time.perf_counter() - t0) * 1000
                            chunks += 1
    except Exception as e:
        err = f"{type(e).__name__}:{str(e)[:120]}"
    total = (time.perf_counter() - t0) * 1000
    out_tok = (usage or {}).get("completion_tokens") or chunks
    return {"idx": idx, "ttft_ms": ttft, "total_ms": total, "output_tokens": out_tok,
            "chunks": chunks, "err": err, "headers": hdrs if idx == 0 else None}

async def run_once(url, headers, model, prompts, n, conc, max_tokens, extra):
    sem = asyncio.Semaphore(conc)
    limits = httpx.Limits(max_connections=conc + 8, max_keepalive_connections=conc + 8)
    async with httpx.AsyncClient(timeout=httpx.Timeout(300.0, connect=30.0), limits=limits) as client:
        async def guarded(i):
            async with sem:
                body = {"model": model, "messages": [{"role": "user", "content": prompts[i % len(prompts)]}],
                        "max_tokens": max_tokens, "temperature": 0, "stream": True,
                        "stream_options": {"include_usage": True}}
                body.update(extra)
                return await one(client, url, headers, body, i)
        t0 = time.perf_counter()
        rows = await asyncio.gather(*[guarded(i) for i in range(n)])
        wall = time.perf_counter() - t0
    return rows, wall

def summarize(rows, wall):
    ok = [r for r in rows if not r["err"] and r["ttft_ms"] is not None]
    bad = [r for r in rows if r["err"] or r["ttft_ms"] is None]
    ttfts = [r["ttft_ms"] for r in ok]; totals = [r["total_ms"] for r in ok]
    per_stream = [r["output_tokens"] / max(1e-6, (r["total_ms"] - r["ttft_ms"]) / 1000) for r in ok if r["output_tokens"] and r["total_ms"] > r["ttft_ms"]]
    return {"n": len(rows), "ok": len(ok), "errors": len(bad), "error_rate": len(bad) / len(rows) if rows else None,
            "wall_s": wall, "out_tokens_total": sum(r["output_tokens"] for r in ok),
            "agg_out_tok_s": sum(r["output_tokens"] for r in ok) / wall if wall else None,
            "per_stream_tok_s_median": statistics.median(per_stream) if per_stream else None,
            "ttft_ms_median": statistics.median(ttfts) if ttfts else None, "ttft_ms_p95": pct(ttfts, .95),
            "latency_ms_median": statistics.median(totals) if totals else None, "latency_ms_p95": pct(totals, .95),
            "err_samples": [r["err"] for r in bad][:3]}

async def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base-url", required=True); ap.add_argument("--key-file", required=True)
    ap.add_argument("--model", required=True); ap.add_argument("--label", required=True)
    ap.add_argument("--runs", type=int, default=3); ap.add_argument("--concurrency", type=int, default=64)
    ap.add_argument("--requests", type=int, default=256); ap.add_argument("--max-tokens", type=int, default=128)
    ap.add_argument("--provider", default=None, help="OpenRouter provider order, e.g. 'AtlasCloud'")
    ap.add_argument("--extra", default="{}", help="extra JSON merged into each body")
    ap.add_argument("--prompts", required=True); ap.add_argument("--out", required=True)
    a = ap.parse_args()
    key = open(a.key_file).read().strip()
    prompts = json.load(open(a.prompts))
    url = a.base_url.rstrip("/") + "/chat/completions" if a.base_url.rstrip("/").endswith("/v1") else a.base_url.rstrip("/") + "/v1/chat/completions"
    headers = {"Authorization": f"Bearer {key}", "Content-Type": "application/json"}
    extra = json.loads(a.extra)
    if a.provider: extra["provider"] = {"order": [a.provider], "allow_fallbacks": False}
    print(f"=== {a.label} | {a.model} | c={a.concurrency} n={a.requests} max_tokens={a.max_tokens} runs={a.runs} ===", flush=True)
    print(f"  warmup (c=4, n=8)...", flush=True)
    await run_once(url, headers, a.model, prompts, 8, 4, a.max_tokens, extra)
    runs = []
    for i in range(a.runs):
        rows, wall = await run_once(url, headers, a.model, prompts, a.requests, a.concurrency, a.max_tokens, extra)
        s = summarize(rows, wall); s["run"] = i + 1; runs.append(s)
        print(f"  run {i+1}: agg {s['agg_out_tok_s']:.0f} tok/s | TTFT p50 {s['ttft_ms_median']:.0f}ms p95 {s['ttft_ms_p95']:.0f}ms | "
              f"lat p50 {s['latency_ms_median']/1000:.2f}s p95 {s['latency_ms_p95']/1000:.2f}s | err {s['error_rate']*100:.1f}% | wall {wall:.1f}s"
              if s["ok"] else f"  run {i+1}: ALL FAILED err={s['err_samples']}", flush=True)
        if i == 0 and rows and rows[0].get("headers"): print(f"    provenance headers: {rows[0]['headers']}", flush=True)
        await asyncio.sleep(5)
    def med(field):
        vs = [r[field] for r in runs if r.get(field) is not None]
        return statistics.median(vs) if vs else None
    med_summary = {k: med(k) for k in ("agg_out_tok_s", "per_stream_tok_s_median", "ttft_ms_median", "ttft_ms_p95",
                                       "latency_ms_median", "latency_ms_p95", "error_rate")}
    out = {"label": a.label, "model": a.model, "base_url": a.base_url, "provider": a.provider,
           "config": {"concurrency": a.concurrency, "requests": a.requests, "max_tokens": a.max_tokens,
                      "runs": a.runs, "prompts": len(prompts), "extra": extra},
           "runs": runs, "median_of_runs": med_summary, "ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())}
    json.dump(out, open(a.out, "w"), indent=1)
    m = med_summary
    print(f"  MEDIAN OF {a.runs} RUNS: agg {m['agg_out_tok_s']:.0f} tok/s | per-stream {m['per_stream_tok_s_median']:.1f} tok/s | "
          f"TTFT p50 {m['ttft_ms_median']:.0f}ms p95 {m['ttft_ms_p95']:.0f}ms | lat p50 {m['latency_ms_median']/1000:.2f}s "
          f"p95 {m['latency_ms_p95']/1000:.2f}s | err {m['error_rate']*100:.1f}%" if m["agg_out_tok_s"] else "  MEDIAN: no successful runs", flush=True)
    print(f"  raw -> {a.out}", flush=True)

asyncio.run(main())