"""Public portability adaptation of the recorded coding benchmark. Prompts and throughput formulas retained; server, credentials and output directory configurable. Original SHA256: dca2d3c234730fafadd6acc56a66ceab28a2b04b2c7b90d90edd520f5239a474 """ import argparse import hashlib import json import os from pathlib import Path import time import urllib.request BASE = os.environ.get("QWEN_API_BASE", "http://localhost:8000").rstrip("/") MODEL = os.environ.get("QWEN_MODEL", "qwen3.8-27b") KEY = os.environ.get("QWEN_API_KEY", "") SUITE_ID = "m23-holdout-v2-9f04c2b7" TASKS = [ ( "python_index", "Write a complete Python 3.12 module implementing an incremental suffix-occurrence " "index over integer token chunks. It must answer the longest suffix match in a bounded " "6..12-token window without rescanning all history, support eviction, use typed " "dataclasses, and include pytest tests. Put the literal suite marker " + SUITE_ID + " in a module constant. Return code only.", ), ( "rust_queue", "Write a complete stable-Rust module for a bounded MPMC queue using per-slot atomic " "sequence counters. Include close semantics, try_push, try_pop, timeout wrappers, safe " "Drop, and multithreaded tests. No external crates. Include the literal marker " + SUITE_ID + ". Return code only.", ), ( "typescript_patch", "Write a complete strict TypeScript module that applies transactional text edits. " "Edits use UTF-16 offsets, may arrive unsorted, must reject overlap, preserve an inverse " "patch, and expose compose and rollback operations. Include node:test cases and the " "literal marker " + SUITE_ID + ". Return code only.", ), ( "go_reload", "Write a complete Go package implementing an atomic configuration reloader. Parse JSON, " "validate cross-field invariants, publish immutable snapshots, debounce file events, " "retain the last good snapshot, and expose observable error state. Include table-driven " "tests and literal marker " + SUITE_ID + ". Return code only.", ), ( "cpp_parser", "Write a complete C++20 incremental WebSocket frame parser with bounded buffering, " "fragment reassembly, masking, control-frame validation, protocol errors, and unit tests " "using only the standard library. Include literal marker " + SUITE_ID + ". Return code only.", ), ( "zig_pool", "Write a complete Zig 0.14 module for a generation-counted object pool. It must detect " "stale handles, support O(1) allocate/free, iterate live objects, and include tests for " "wraparound and double free. Include literal marker " + SUITE_ID + ". Return code only.", ), ( "sql_ledger", "Write a PostgreSQL 17 migration and verification script for an append-only double-entry " "ledger. Enforce balanced transactions with deferred constraints, idempotency keys, " "tenant isolation, reversal entries, useful indexes, and pgTAP-style checks. Include " "literal marker " + SUITE_ID + ". Return SQL only.", ), ( "cuda_scan", "Write a complete CUDA C++ program implementing an exclusive prefix sum for arbitrary " "length float arrays using hierarchical block scans. Avoid bank conflicts, handle tails, " "check CUDA errors, compare against a CPU reference, and benchmark. Include literal " "marker " + SUITE_ID + ". Return code only.", ), ] def metrics() -> tuple[float, float]: request = urllib.request.Request( BASE + "/metrics", headers={"Authorization": "Bearer " + KEY} ) values: dict[str, float] = {} for line in urllib.request.urlopen(request, timeout=10).read().decode().splitlines(): for name in ( "vllm:spec_decode_num_drafts_total", "vllm:spec_decode_num_accepted_tokens_total", ): if line.startswith(name + " ") or line.startswith(name + "{"): values[name] = float(line.split()[-1]) return ( values.get("vllm:spec_decode_num_drafts_total", 0.0), values.get("vllm:spec_decode_num_accepted_tokens_total", 0.0), ) def run_request(prompt: str, max_tokens: int) -> dict[str, object]: payload = { "model": MODEL, "messages": [{"role": "user", "content": prompt}], "max_tokens": max_tokens, "temperature": 0, "stream": True, "stream_options": {"include_usage": True}, "chat_template_kwargs": {"enable_thinking": False}, } request = urllib.request.Request( BASE + "/v1/chat/completions", data=json.dumps(payload).encode(), headers={ "Content-Type": "application/json", "Authorization": "Bearer " + KEY, }, ) drafts0, accepted0 = metrics() started = time.perf_counter() first: float | None = None usage: dict[str, int] = {} chunks: list[str] = [] with urllib.request.urlopen(request, timeout=1800) as response: for raw in response: line = raw.decode().strip() if not line.startswith("data: "): continue body = line[6:] if body == "[DONE]": break event = json.loads(body) if event.get("usage"): usage = event["usage"] for choice in event.get("choices", []): content = choice.get("delta", {}).get("content") if content: first = first or time.perf_counter() chunks.append(content) ended = time.perf_counter() drafts1, accepted1 = metrics() output = "".join(chunks) output_tokens = int(usage.get("completion_tokens", len(chunks))) steps = drafts1 - drafts0 accepted = accepted1 - accepted0 decode_seconds = max(ended - (first or ended), 1e-6) return { "prompt_tokens": int(usage.get("prompt_tokens", 0)), "output_tokens": output_tokens, "steps": steps, "accepted": accepted, "tokens_per_step": 1.0 + accepted / max(steps, 1.0), "decode_tps": max(output_tokens - 1, 0) / decode_seconds, "ttft": (first or ended) - started, "sha256": hashlib.sha256(output.encode()).hexdigest()[:16], } def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("tag") parser.add_argument("--max-tokens", type=int, default=384) args = parser.parse_args() # Warm the exact graph length without reusing any benchmark prompt. run_request( "Return a Python function that adds two integers. Marker warmup-" + SUITE_ID, 96, ) rows: list[dict[str, object]] = [] for name, prompt in TASKS: row = {"task": name, **run_request(prompt, args.max_tokens)} rows.append(row) print( f"HOLDOUT {args.tag} {name:16s} out={row['output_tokens']:4d} " f"steps={row['steps']:5.0f} tok/step={row['tokens_per_step']:5.2f} " f"decode={row['decode_tps']:6.1f} ttft={row['ttft']:5.2f}s " f"sha={row['sha256']}", flush=True, ) total_out = sum(int(row["output_tokens"]) for row in rows) total_steps = sum(float(row["steps"]) for row in rows) total_accepted = sum(float(row["accepted"]) for row in rows) total_time = sum( max(int(row["output_tokens"]) - 1, 0) / max(float(row["decode_tps"]), 1e-9) for row in rows ) summary = { "suite_id": SUITE_ID, "tag": args.tag, "rows": rows, "total_tokens_per_step": 1.0 + total_accepted / max(total_steps, 1.0), "total_decode_tps": max(total_out - len(rows), 0) / max(total_time, 1e-9), } print( f"HOLDOUT {args.tag} TOTAL out={total_out} steps={total_steps:.0f} " f"tok/step={summary['total_tokens_per_step']:.2f} " f"decode={summary['total_decode_tps']:.1f} tok/s", flush=True, ) output_dir = Path(os.environ.get("QWEN_RESULTS_DIR", "./results")) output_dir.mkdir(parents=True, exist_ok=True) (output_dir / f"coding_holdout_v2_{args.tag}.json").write_text( json.dumps(summary, indent=2) + "\n" ) if __name__ == "__main__": main()