']
for label, frac, val, cls in rows:
out.append(f'{escape(label)}'
f'{val}')
return "".join(out) + "
"
def best_speedup(ep):
return max((t["speedup"] for t in ep["turns"] if t["correct"]), default=None)
def furthest(ep):
return max((STAGE_ORDER.index(t["stage"]) for t in ep["turns"]), default=0)
def diff_html(prev, cur):
lines = list(difflib.unified_diff(prev.splitlines(), cur.splitlines(), "previous", "this turn", lineterm="", n=2))
if not lines:
return '
Identical to the previous submission.
'
cls = lambda ln: "d-hunk" if ln.startswith("@@") else "d-add" if ln.startswith("+") else "d-del" if ln.startswith("-") else ""
body = "".join(f'{escape(ln) or " "}' if cls(ln) else escape(ln) + "\n" for ln in lines[2:])
return f"
{body}
"
def episode_html(ep):
hist = ep.get("history") or []
turns_html, prev_code = [], None
for k, t in enumerate(ep["turns"]):
h = hist[k] if k < len(hist) else {}
completion = ep["completions"][k] if k < len(ep.get("completions", [])) else ""
code = extract_code(completion)
bits = [f"Turn {k + 1}", chip(t["stage"])]
if t["correct"]:
bits.append(f'speedup {t["speedup"]:.2f}×')
if t.get("t_ref_ms") is not None:
bits.append(f'ref {fmt(t["t_ref_ms"], ".3f", " ms")} → yours {fmt(t.get("t_new_ms"), ".3f", " ms")}')
if t.get("t_min_ms") is not None:
bits.append(f'{escape(t.get("bound") or "")}-bound, floor {t["t_min_ms"]:.3f} ms, '
f'roofline {fmt(t.get("roofline_frac"), ".0%")} (PyTorch {fmt(t.get("ref_roofline_frac"), ".0%")})')
if t.get("truncated"):
bits.append(f'cut off at token limit')
if t.get("completion_tokens"):
bits.append(f'{t["completion_tokens"]} tokens')
if h.get("roofline_violation"):
bits.append('below hardware floor')
if h.get("eval_s") is not None:
bits.append(f'eval {h["eval_s"]:.1f} s, GPU queue {h.get("queue_wait_s") or 0:.1f} s')
if t.get("gen_s") is not None:
bits.append(f'generation {t["gen_s"]:.0f} s')
if h.get("suspicious"):
bits.append('suspicious speedup')
parts = [f'
')
if prev_code is not None:
parts.append(f'Diff vs turn {k}{diff_html(prev_code, code)}')
parts.append(f'Raw completion ({len(completion)} chars)'
f'
{escape(completion)}
')
turns_html.append(f'
{"".join(parts)}
')
prev_code = code
best = best_speedup(ep)
eid = f'episode {escape(ep["episode_id"])}' if ep.get("episode_id") else ""
if ep.get("error"):
eid += f'
Episode error: {escape(ep["error"])}
'
return (f'{escape(ep["task_id"])}'
f'{"".join(chip(t["stage"]) for t in ep["turns"])}'
f' 1 else ""}">{fmt(best, ".2f", "×")}'
f'
{eid}
{"".join(turns_html)}')
def render(eps, title):
n = len(eps)
s = summarize(eps)
fur = [furthest(ep) for ep in eps]
bests = [best_speedup(ep) for ep in eps]
idx = STAGE_ORDER.index
stats = [("episodes", str(n)), ("compiled", f'{s["compile_pct"]:.0f}%'), ("correct", f'{s["correct_pct"]:.0f}%'),
("correct on turn 1", f'{s["correct_turn1_pct"]:.0f}%'), ("faster than PyTorch", f'{s["faster_than_ref_pct"]:.0f}%'),
("median speedup", fmt(s["median_speedup"], ".2f", "×")),
("median roofline", fmt(s.get("median_roofline_frac"), ".0%")), ("suspicious", str(s["suspicious"]))]
if s.get("samples_per_task"): # several samples per task: pass@1 / pass@k over tasks
k = s["samples_per_task"]
stats[:1] = [("tasks × samples", f'{s["tasks"]} × {k}'), ("pass@1", f'{s["pass@1_pct"]:.0f}%'),
(f"pass@{k}", f'{s[f"pass@{k}_pct"]:.0f}%')]
funnel_counts = [
("Submitted", n),
("Passed code checks", sum(f >= idx("compile") for f in fur)),
("Compiled", sum(f >= idx("runtime") for f in fur)),
("Ran to completion", sum(f >= idx("correctness") for f in fur)),
("Correct", sum(f == idx("timed") for f in fur)),
("Faster than PyTorch", sum(b is not None and b > 1 for b in bests)),
]
funnel = bar_rows([(lab, c / n if n else 0, f"{c} · {pct(c, n):.0f}%", "") for lab, c in funnel_counts])
max_turns = max((len(ep["turns"]) for ep in eps), default=0)
turn_rows = []
for k in range(max_turns):
at = [ep["turns"][k] for ep in eps if len(ep["turns"]) > k]
cnt = Counter(t["stage"] for t in at)
cum = sum(any(t["correct"] for t in ep["turns"][:k + 1]) for ep in eps)
stack = "".join(f'' for st in STAGE_ORDER if cnt[st])
gen = next((t.get("gen_s") for t in at if t.get("gen_s") is not None), None)
turn_rows.append(f'
{k + 1}
{len(at)}
'
f'
{stack}
'
f'
{pct(cum, n):.0f}%
{fmt(gen, ".0f", " s")}
')
legend = "".join(f'{STAGE_LABEL[st]}' for st in STAGE_ORDER)
sp = [b for b in bests if b is not None]
buckets = [(lab, sum(lo <= b < hi for b in sp)) for lo, hi, lab in SPEEDUP_BUCKETS]
top = max((c for _, c in buckets), default=0) or 1
speed = (bar_rows([(lab, c / top, str(c), "ok" if lo >= 1 else "warn") for (lab, c), (lo, _, _) in zip(buckets, SPEEDUP_BUCKETS)])
if sp else '
No correct submissions yet.
')
hist = [h for ep in eps for h in ep.get("history") or [] if h.get("eval_s") is not None]
gpu_hist = [h for h in hist if h.get("gpu_id") is not None]
steps = [t.get("step_s") for ep in eps for t in ep["turns"]]
timing_rows = [
("Generation per turn", [t.get("gen_s") for ep in eps for t in ep["turns"]]),
("Env step, client side", steps),
("GPU evaluation", [h["eval_s"] for h in gpu_hist]),
("Waiting for a free GPU", [h.get("queue_wait_s") for h in gpu_hist]),
]
timing = "".join(f'
{lab}
{fmt(quantile(v, .5), ".1f", " s")}
'
f'
{fmt(quantile(v, .95), ".1f", " s")}
{fmt(max((x for x in v if x is not None), default=None), ".1f", " s")}
'
for lab, v in timing_rows if any(x is not None for x in v))
timing = (f'
Phase
p50
p95
max
{timing}
'
if timing else '
No timing fields in this file (written before timing was logged).
')
# roofline: PyTorch vs best correct kernel per task, as share of the hardware floor
roof_rows = []
for ep in eps:
t_best = max((t for t in ep["turns"] if t.get("ref_roofline_frac") is not None),
key=lambda t: (t["correct"], t.get("roofline_frac") or 0), default=None)
if t_best:
mine = t_best.get("roofline_frac") if t_best["correct"] else None
roof_rows.append(f'
{escape(ep["task_id"])}
{escape(t_best.get("bound") or "–")}
'
f'
{fmt(t_best.get("t_min_ms"), ".3f", " ms")}
'
f'
{fmt(t_best["ref_roofline_frac"], ".0%")}
'
f'
{fmt(mine, ".0%")}
'
f'
{fmt(1 / t_best["ref_roofline_frac"] if t_best["ref_roofline_frac"] else None, ".1f", "×")}
')
roof_html = (f'
Task
Bound
Floor
PyTorch
Best kernel
'
f'
Max speedup
{"".join(roof_rows)}
Share of the roofline floor '
f'(max of FLOPs ÷ peak compute, bytes ÷ peak bandwidth). Max speedup = how much faster than PyTorch any kernel could be.
'
if roof_rows else '
No roofline data (file predates it, or the GPU is not in the peak table).
')
order = sorted(eps, key=lambda ep: (-furthest(ep), -(best_speedup(ep) or 0), ep["task_id"]))
stat_html = "".join(f'
{v}{lab}
' for lab, v in stats)
return f"""{escape(title)}
kernel_env run report
{escape(title)}
{n} KernelBench episodes, up to {max_turns} submissions each. Speedup is eager PyTorch time divided by kernel time on the eval GPU.
{stat_html}
Where episodes stopped
{funnel}
Best stage each episode reached in any turn.
Best speedup, correct episodes
{speed}
Turn by turn
Turn
Submissions
Outcome mix
Correct so far
Generation
{"".join(turn_rows)}
{legend}
Distance to the hardware roof
{roof_html}
Where the time went
{timing}
Episodes
Sorted by furthest stage, then speedup. Open one to see each submission, its feedback and the diff from the previous turn.
{"".join(episode_html(ep) for ep in order)}"""
def main():
p = argparse.ArgumentParser()
p.add_argument("episodes", help="JSONL from baseline_eval.py --out")
p.add_argument("-o", "--out", default="report.html")
p.add_argument("--title", default=None)
args = p.parse_args()
eps = [json.loads(line) for line in Path(args.episodes).read_text().splitlines() if line.strip()]
Path(args.out).write_text(render(eps, args.title or Path(args.episodes).stem))
print(f"{len(eps)} episodes -> {args.out}")
if __name__ == "__main__":
main()