Use Pareto as an LLM judge to grade model outputs

"""Pairwise judge with both positions. Code: MIT. No candidate is from Pareto."""
import json
import os

MODEL = os.environ.get('MODEL', 'pareto')
BASE_URL = os.environ.get('BASE_URL', 'https://api.unbiased.ai/v1')


def request(item, config, swapped=False):
    a, b = item['answer_a'], item['answer_b']
    if swapped:
        a, b = b, a
    return {'messages': [
        {'role': 'system', 'content': config['prompt']},
        {'role': 'user', 'content': json.dumps({'question': item['text'], 'A': a, 'B': b}, ensure_ascii=False)}],
        'tools': [{'type': 'function', 'function': {'name': 'grade', 'description': 'Record the preferred answer and a short reason.',
            'parameters': {'type': 'object', 'properties': {'choice': {'type': 'string', 'enum': ['A', 'B', 'equal']},
                'reason': {'type': 'string'}}, 'required': ['choice', 'reason'], 'additionalProperties': False}}}],
        'tool_choice': config['tool_choice'], 'parallel_tool_calls': False,
        'max_tokens': config['max_tokens'], 'reasoning_effort': config['reasoning_effort']}


def plan_requests(item, config):
    return [request(item, config), request(item, config, True)]


def parse(response):
    calls = response['choices'][0]['message'].get('tool_calls') or []
    if len(calls) != 1 or calls[0]['function']['name'] != 'grade':
        return {'choice': 'invalid', 'reason': 'Expected exactly one grade call.'}
    try:
        args = json.loads(calls[0]['function']['arguments'])
        if set(args) != {'choice', 'reason'} or args['choice'] not in ('A', 'B', 'equal') or not isinstance(args['reason'], str):
            raise ValueError('invalid schema')
        return args
    except (ValueError, TypeError):
        return {'choice': 'invalid', 'reason': 'Invalid grade arguments.'}


def generate(item, config, call):
    return {'original': parse(call(request(item, config))),
            'swapped': parse(call(request(item, config, True)))}

"""Offline pairwise checker. Code: MIT; never executes model text."""
import argparse
import gzip
import json
from pathlib import Path


def score(item, arguments, config):
    original = arguments.get('original', {}).get('choice', 'invalid')
    swapped = arguments.get('swapped', {}).get('choice', 'invalid')
    allowed = ('A', 'B', 'equal', 'invalid')
    if original not in allowed:
        original = 'invalid'
    if swapped not in allowed:
        swapped = 'invalid'
    normalized = {'A': 'B', 'B': 'A'}.get(swapped, swapped)
    return {'gold': item['gold'], 'prediction': original,
            'secondary': {'agreement': {'gold': item['gold'], 'prediction': original},
                          'position_consistency': {'score': int(original != 'invalid' and original == normalized)},
                          'valid_output': {'score': int(original != 'invalid' and swapped != 'invalid')}}}


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='Check scored records against fixture labels; no API calls.')
    parser.add_argument('raw', help='raw.jsonl.gz or scored.jsonl.gz')
    parser.add_argument('--fixture', default=str(Path(__file__).with_name('fixture.jsonl')))
    args = parser.parse_args()
    items = {r['item_id']: r for r in map(json.loads, Path(args.fixture).read_text().splitlines())}
    opener = gzip.open if args.raw.endswith('.gz') else open
    mismatches = 0
    with opener(args.raw, 'rt') as stream:
        for line in stream:
            row = json.loads(line)
            expected = score(items[row['item_id']], row['tool_arguments'], {})
            if expected['prediction'] != expected['gold']:
                mismatches += 1
                print(row['item_id'], 'repeat', row['repeat'], 'expected', expected['gold'], 'got', expected['prediction'])
            for key in expected:
                if key in row and row[key] != expected[key]:
                    raise SystemExit('Stored score differs from checker')
    print('Verified stored scores; label mismatches:', mismatches)

#!/usr/bin/env python3
"""Generated eval runner. Code: MIT. Requires Python 3.10+; --plan needs tiktoken==0.14.0."""
import sys, types
from pathlib import Path
runtime = types.ModuleType("runtime")
sys.modules["runtime"] = runtime
exec(compile('"""Key-free eval primitives. Only allowlisted records may leave this module."""\nimport email.utils\nimport gzip\nimport hashlib\nimport types\nimport json\nimport math\nimport os\nfrom pathlib import Path\nimport re\nimport shutil\nimport time\nimport urllib.error\nimport urllib.request\nfrom datetime import datetime, timezone\n\nPARETO_URL = "https://api.unbiased.ai/v1"\nBASELINE_URL = "https://api.openai.com/v1"\nBASELINE_MODEL = "gpt-6.1-sol"\nMAX_RAW_BYTES = 64 * 1024 * 1024\nRAW_KEYS = frozenset("item_id arm repeat request_model response_model endpoint provider_served allow_fallbacks reasoning_effort max_tokens tool_choice tool_arguments finish_reason tokens response_id started_at errors invalid_tool_calls retries gold prediction score tp fp fn secondary latency_ms".split())\nSCORE_KEYS = frozenset("gold prediction score tp fp fn secondary".split())\nTOKEN_KEYS = ("input", "output", "cached", "reasoning")\nSECRET = re.compile(r"\\b(?:sk-(?:proj-|or-v1-)?[A-Za-z0-9_-]{16,}|xox[bcp]-[A-Za-z0-9-]{16,})")\n\n\ndef sha256(data):\n    return hashlib.sha256(data).hexdigest()\n\n\ndef json_bytes(value):\n    return (json.dumps(value, ensure_ascii=False, indent=2) + "\\n").encode()\n\n\ndef jsonl_bytes(rows):\n    return ("".join(json.dumps(r, ensure_ascii=False, separators=(",", ":"), allow_nan=False) + "\\n" for r in rows)).encode()\n\n\ndef read_rows(path):\n    data = Path(path).read_bytes()\n    if str(path).endswith(".gz"):\n        with gzip.open(path, "rb") as stream:\n            data = stream.read(MAX_RAW_BYTES + 1)\n    if len(data) > MAX_RAW_BYTES:\n        raise ValueError("raw/fixture exceeds size limit")\n    return [json.loads(line) for line in data.decode().splitlines() if line.strip()]\n\n\ndef approved_raw(row):\n    if set(row) - RAW_KEYS:\n        raise ValueError("raw contains unapproved fields")\n    if set(row.get("tokens", {})) != set(TOKEN_KEYS):\n        raise ValueError("raw usage shape mismatch")\n    t = row["tokens"]\n    if any(type(t[k]) is not int or t[k] < 0 for k in TOKEN_KEYS) or t["cached"] > t["input"] or t["reasoning"] > t["output"]:\n        raise ValueError("invalid usage tokens")\n    encoded = json.dumps(row, ensure_ascii=False, allow_nan=False)\n    if SECRET.search(encoded):\n        raise ValueError("raw contains a key-like string")\n    return row\n\n\nclass Limiter:\n    def __init__(self, rpm=18, clock=time.monotonic, sleep=time.sleep):\n        if not math.isfinite(rpm) or not 0 < rpm <= 30:\n            raise ValueError("rpm must be positive and at most 30")\n        self.interval, self.clock, self.sleep = 60 / rpm, clock, sleep\n        self.next = 0\n\n    def wait(self):\n        delay = self.next - self.clock()\n        if delay > 0:\n            self.sleep(delay)\n        self.next = self.clock() + self.interval\n\n\ndef retry_after(value, now=None):\n    try:\n        return max(0, float(value))\n    except (TypeError, ValueError):\n        try:\n            when = email.utils.parsedate_to_datetime(value).timestamp()\n            return max(0, when - (time.time() if now is None else now))\n        except (TypeError, ValueError, OverflowError):\n            return None\n\n\ndef identity(arm, model, endpoint, response=None):\n    if arm == "baseline":\n        if model != BASELINE_MODEL or endpoint != BASELINE_URL:\n            raise ValueError("baseline request not allowlisted; OpenRouter is unverified and disabled")\n        if response is not None:\n            if not re.fullmatch(r"gpt-6\\.1-sol(?:-\\d{4}-\\d{2}-\\d{2})?", response.get("model", "")):\n                raise ValueError("baseline response model not allowlisted")\n            # Direct OpenAI is the provider attestation. A contradictory field\n            # still aborts, rather than silently relabeling a proxied response.\n            if response.get("provider", "OpenAI") != "OpenAI":\n                raise ValueError("baseline provider not OpenAI")\n    elif arm == "pareto":\n        if model != "pareto" or endpoint != PARETO_URL:\n            raise ValueError("Pareto request not allowlisted")\n        if response is not None and not re.fullmatch(r"pareto-\\d{2}\\.\\d{1,2}", response.get("model", "")):\n            raise ValueError("Pareto response release missing")\n    else:\n        raise ValueError("unknown arm")\n\n\ndef api_key(arm):\n    name = "UNBIASED_API_KEY" if arm == "pareto" else "OPENAI_API_KEY"\n    value = os.environ.get(name)\n    if not value and arm == "pareto":\n        path = Path.home() / ".claude/secrets/unbiased-api.key"\n        if path.is_file():\n            value = path.read_text().strip()\n    if not value:\n        raise ValueError("missing " + name)\n    return value\n\n\nclass NoRedirect(urllib.request.HTTPRedirectHandler):\n    def redirect_request(self, req, fp, code, msg, headers, newurl):\n        raise ValueError("API redirects are forbidden")\n\n\nclass Client:\n    def __init__(self, arm, model, endpoint, key, limiter, opener=None, sleep=time.sleep, max_retries=3):\n        identity(arm, model, endpoint)\n        self.arm, self.model, self.endpoint = arm, model, endpoint\n        self.key, self.limiter, self.sleep, self.max_retries = key, limiter, sleep, max_retries\n        self.opener = opener or urllib.request.build_opener(NoRedirect()).open\n\n    def call(self, payload):\n        allowed = {"model", "messages", "tools", "tool_choice", "max_tokens", "reasoning_effort", "parallel_tool_calls"}\n        if set(payload) - allowed:\n            raise ValueError("request contains unapproved fields")\n        if any(not isinstance(m, dict) or m.get("content") is not None and not isinstance(m["content"], str) for m in payload.get("messages", [])):\n            raise ValueError("only text messages are supported")\n        encoded = json.dumps(payload, ensure_ascii=False)\n        if self.key in encoded or SECRET.search(encoded):\n            raise ValueError("request contains a key-like string")\n        if payload.get("model", self.model) != self.model:\n            raise ValueError("recipe changed model")\n        body = dict(payload, model=self.model)\n        retries = 0\n        while True:\n            self.limiter.wait()\n            req = urllib.request.Request(self.endpoint + "/chat/completions", data=json_bytes(body), headers={"Authorization": "Bearer " + self.key, "Content-Type": "application/json"})\n            try:\n                with self.opener(req, timeout=120) as resp:\n                    data = resp.read(MAX_RAW_BYTES + 1)\n                    if len(data) > MAX_RAW_BYTES:\n                        raise ValueError("API response exceeds size limit")\n                    response = json.loads(data)\n                identity(self.arm, self.model, self.endpoint, response)\n                output = json.dumps(response, ensure_ascii=False)\n                if self.key in output or SECRET.search(output):\n                    raise ValueError("model output contains a key-like string")\n                return response, retries\n            except urllib.error.HTTPError as error:\n                if error.code not in (429, 500, 502, 503, 504) or retries >= self.max_retries:\n                    raise ValueError("API request failed (HTTP %s); body withheld" % error.code) from None\n                delay = retry_after(error.headers.get("Retry-After"))\n                error.close()\n                self.sleep(delay if delay is not None else min(60, 2 ** retries))\n                retries += 1\n            except (urllib.error.URLError, TimeoutError):\n                # Do not retry an ambiguous completed request automatically.\n                raise ValueError("API transport failed; inspect partial run before resuming") from None\n\n\ndef usage(response):\n    u = response.get("usage", {})\n    return {"input": u.get("prompt_tokens", 0), "output": u.get("completion_tokens", 0),\n            "cached": (u.get("prompt_tokens_details") or {}).get("cached_tokens", 0),\n            "reasoning": (u.get("completion_tokens_details") or {}).get("reasoning_tokens", 0)}\n\n\ndef score_record(item, arguments, checker, config):\n    scored = checker.score(item, arguments, config)\n    if not isinstance(scored, dict) or set(scored) - SCORE_KEYS:\n        raise ValueError("checker returned unapproved fields")\n    shape = {"majority_label": ("gold", "prediction"), "median": ("score",), "summed_counts": ("tp", "fp", "fn")}\n    def check(payload, aggregation):\n        required = shape[aggregation]\n        if set(payload) != set(required):\n            raise ValueError("checker record shape mismatch")\n        if aggregation == "median" and (type(payload["score"]) not in (int, float) or not math.isfinite(payload["score"])):\n            raise ValueError("checker score must be finite")\n        if aggregation == "summed_counts" and any(type(payload[k]) is not int or payload[k] < 0 for k in required):\n            raise ValueError("checker counts must be nonnegative integers")\n    check({k: v for k, v in scored.items() if k != "secondary"}, config["aggregation"])\n    if config["aggregation"] == "majority_label" and scored["gold"] != item["gold"]:\n        raise ValueError("checker changed frozen gold")\n    secondary = scored.setdefault("secondary", {})\n    if set(secondary) != {s["id"] for s in config.get("secondary", [])}:\n        raise ValueError("checker secondary metrics mismatch")\n    for spec in config.get("secondary", []):\n        check(secondary[spec["id"]], spec["aggregation"])\n    return scored\n\n\ndef generate_record(item, arm, repeat, config, recipe, client):\n    started = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")\n    begin = time.monotonic()\n    calls = []\n    def call(payload):\n        if payload.get("max_tokens") != config["max_tokens"] or payload.get("reasoning_effort") != config["reasoning_effort"]:\n            raise ValueError("recipe changed frozen token limit or reasoning effort")\n        choices = [config["tool_choice"], *config.get("tool_steps", [])]\n        if payload.get("tool_choice") not in choices or payload.get("tool_choice", {}).get("type") != "function":\n            raise ValueError("recipe changed frozen forced tool choice")\n        if not isinstance(payload.get("tools"), list) or not payload["tools"]:\n            raise ValueError("recipe requires tools")\n        response, retries = client.call(payload)\n        calls.append((response, retries, payload["tool_choice"]))\n        return response\n    # generate may implement a dependent tool loop using this checked call.\n    combined = None\n    if hasattr(recipe, "generate"):\n        combined = recipe.generate(item, config, call)\n    else:\n        call(recipe.request(item, config))\n    if not calls:\n        raise ValueError("recipe did not call the model")\n    response = calls[-1][0]\n    final = calls[-1][2]\n    if final != config["tool_choice"]:\n        raise ValueError("loop must end on the frozen forced tool choice")\n    choice = response["choices"][0]\n    tools = choice.get("message", {}).get("tool_calls", [])\n    invalid = 0\n    arguments = {}\n    try:\n        if len(tools) != 1 or tools[0]["function"]["name"] != final["function"]["name"]:\n            raise ValueError("expected exactly one final named tool call")\n        arguments = json.loads(tools[0]["function"]["arguments"])\n        if not isinstance(arguments, dict):\n            raise ValueError("tool arguments must be an object")\n    except (ValueError, KeyError, TypeError):\n        arguments, invalid = {}, 1\n    if combined is not None:\n        if not isinstance(combined, dict):\n            raise ValueError("recipe combined arguments must be an object")\n        arguments = combined\n    models = {r["model"] for r, _, _ in calls}\n    if len(models) != 1:\n        raise ValueError("served model changed within item")\n    row = {"item_id": item["item_id"], "arm": arm, "repeat": repeat,\n           "request_model": client.model, "response_model": response["model"], "endpoint": client.endpoint,\n           "provider_served": "OpenAI" if arm == "baseline" else "Unbiased", "allow_fallbacks": False,\n           "reasoning_effort": config["reasoning_effort"], "max_tokens": config["max_tokens"], "tool_choice": config["tool_choice"],\n           "tool_arguments": arguments, "finish_reason": "length" if any(r["choices"][0]["finish_reason"] == "length" for r, _, _ in calls) else choice["finish_reason"],\n           "tokens": {k: sum(usage(r)[k] for r, _, _ in calls) for k in TOKEN_KEYS},\n           "response_id": ",".join(r["id"] for r, _, _ in calls), "started_at": started, "errors": 0,\n           "invalid_tool_calls": invalid, "retries": sum(n for _, n, _ in calls), "latency_ms": round((time.monotonic() - begin) * 1000)}\n    return approved_raw(row)\n\n\ndef load_module(path, name):\n    module = types.ModuleType(name)\n    module.__file__ = str(path)\n    exec(compile(Path(path).read_bytes(), str(path), "exec"), module.__dict__)\n    return module\n\n\ndef retain_runs(directory):\n    # Only directories created by this runner (with its marker) are eligible.\n    runs = sorted(p for p in directory.iterdir() if p.is_dir() and not p.is_symlink() and (p / "manifest.json").is_file())\n    for old in runs[:-2]:\n        manifest = json.loads((old / "manifest.json").read_text())\n        if manifest.get("format") != "uc-eval-run-v1":\n            raise ValueError("refusing to remove unrecognized run")\n        if set(p.name for p in old.iterdir()) - {"manifest.json", "raw.jsonl.gz", "scored.jsonl.gz", "receipt.json"}:\n            raise ValueError("refusing to remove run with unexpected files")\n        shutil.rmtree(old)\n\n\ndef write_raw(path, rows):\n    data = jsonl_bytes([approved_raw(row) for row in rows])\n    if len(data) > MAX_RAW_BYTES:\n        raise ValueError("raw log exceeds size limit")\n    Path(path).write_bytes(gzip.compress(data, mtime=0))\n', "runtime", "exec"), runtime.__dict__)
sandbox = types.ModuleType("sandbox")
sys.modules["sandbox"] = sandbox
exec(compile('"""Local macOS checker sandbox. No keys, network, child processes or host writes."""\nimport json\nimport os\nfrom pathlib import Path\nimport signal\nimport subprocess\nimport sys\nimport tempfile\nimport time\n\n\n# This worker is trusted code. Generated Python may execute only after the\n# sandbox and rlimits are active. Checker modules must use in-process unittest,\n# not subprocesses. Each score gets a fresh interpreter and empty working dir.\nWORKER = r\'\'\'\nimport json, resource, sys, types\nfrom pathlib import Path\ncpu, memory = int(sys.argv[1]), int(sys.argv[2])\nresource.setrlimit(resource.RLIMIT_CPU, (cpu, cpu))\n# macOS rejects a small RLIMIT_DATA for modern Python. RSS is supervised\n# by the parent; CPU/file/core/descriptor limits are still enforced here.\nresource.setrlimit(resource.RLIMIT_FSIZE, (1048576, 1048576))\nresource.setrlimit(resource.RLIMIT_CORE, (0, 0))\nresource.setrlimit(resource.RLIMIT_NOFILE, (64, 64))\npayload = json.loads(Path(\'input.json\').read_text())\nchecker = types.ModuleType(\'uc_checker\')\nchecker.__file__ = str(Path(\'check.py\').resolve())\nexec(compile(Path(\'check.py\').read_bytes(), \'check.py\', \'exec\'), checker.__dict__)\nresult = checker.score(payload[\'item\'], payload[\'arguments\'], payload[\'config\'])\nPath(\'result.json\').write_text(json.dumps(result, allow_nan=False))\n\'\'\'\n\n\ndef score(checker_path, item, arguments, config, *, cpu_seconds=5, memory_mb=256, wall_seconds=10):\n    if sys.platform != \'darwin\' or not Path(\'/usr/bin/sandbox-exec\').is_file():\n        raise RuntimeError(\'macOS sandbox-exec is required; no unsandboxed fallback\')\n    if cpu_seconds < 1 or memory_mb < 16 or wall_seconds <= 0:\n        raise ValueError(\'invalid sandbox limits\')\n    memory = memory_mb * 1024 * 1024\n    with tempfile.TemporaryDirectory(prefix=\'uc-sandbox-\') as directory:\n        work = Path(directory).resolve()\n        (work / \'check.py\').write_bytes(Path(checker_path).read_bytes())\n        (work / \'input.json\').write_text(json.dumps({\'item\': item, \'arguments\': arguments, \'config\': config}))\n        # No blanket file-read: this machine has credentials. Python and OS\n        # libraries are read-only; only this fresh directory is writable.\n        roots = sorted({str(Path(sys.base_prefix).resolve()), sys.base_prefix, \'/System/Library\', \'/usr/lib\', \'/Library/Apple\', \'/private/var/db/dyld\'})\n        profile = \'(version 1)(deny default)(allow process-exec)(allow sysctl-read)\'\n        profile += \'(allow file-read-metadata)(allow file-read* (literal "/"))\'\n        profile += \'(allow file-read* \' + \' \'.join(\'(subpath \' + json.dumps(p) + \')\' for p in roots + [str(work)]) + \')\'\n        profile += \'(allow file-read* (literal \' + json.dumps(str(Path(sys.executable).resolve())) + \') (literal "/dev/urandom") (literal "/dev/null"))\'\n        profile += \'(allow file-write* (subpath \' + json.dumps(str(work)) + \') (literal "/dev/null"))\'\n        # Default-deny includes network*, process-fork, mach lookup and host IO.\n        env = {\'PATH\': \'/usr/bin:/bin\', \'HOME\': str(work), \'TMPDIR\': str(work), \'LC_ALL\': \'C\', \'PYTHONDONTWRITEBYTECODE\': \'1\'}\n        command = [\'/usr/bin/sandbox-exec\', \'-p\', profile, str(Path(sys.executable).resolve()), \'-I\', \'-B\', \'-c\', WORKER, str(cpu_seconds), str(memory)]\n        proc = subprocess.Popen(command, cwd=work, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)\n        deadline = time.monotonic() + wall_seconds\n        try:\n            while proc.poll() is None:\n                if time.monotonic() >= deadline:\n                    raise TimeoutError(\'sandbox wall-time limit\')\n                # macOS does not enforce RLIMIT_RSS; supervise resident memory\n                # too. Fork is denied, so there can be no uncounted descendants.\n                rss = subprocess.run([\'/bin/ps\', \'-o\', \'rss=\', \'-p\', str(proc.pid)], capture_output=True, text=True, timeout=1)\n                if rss.returncode == 0 and rss.stdout.strip() and int(rss.stdout.strip()) * 1024 > memory:\n                    raise MemoryError(\'sandbox resident-memory limit\')\n                time.sleep(0.02)\n            if proc.returncode != 0:\n                raise RuntimeError(\'sandbox checker failed\')\n            result = work / \'result.json\'\n            if result.is_symlink() or result.stat().st_size > 1048576:\n                raise ValueError(\'invalid sandbox result\')\n            return json.loads(result.read_text())\n        finally:\n            if proc.poll() is None:\n                os.killpg(proc.pid, signal.SIGKILL)\n            proc.wait(timeout=2)\n', "sandbox", "exec"), sandbox.__dict__)
app = types.ModuleType("uc_app")
exec(compile('"""CLI shared by the private runner and generated per-page downloads."""\nimport argparse\nfrom datetime import datetime, timezone\nimport json\nimport math\nimport os\nfrom pathlib import Path\nimport re\nimport subprocess\nimport sys\n\nfrom runtime import (BASELINE_MODEL, BASELINE_URL, PARETO_URL, SCORE_KEYS, TOKEN_KEYS,\n                     Client, Limiter, api_key, approved_raw, generate_record, identity,\n                     json_bytes, load_module, read_rows, retain_runs, score_record,\n                     sha256, write_raw)\n\n\ndef canon(root):\n    script = "import {ACQUISITION_PLANS} from \'./scripts/claims.mjs\';import {readFileSync} from \'node:fs\';console.log(JSON.stringify({personal:ACQUISITION_PLANS.personal,prices:JSON.parse(readFileSync(\'data/prices.json\')).models}))"\n    return json.loads(subprocess.check_output(["node", "--input-type=module", "-e", script], cwd=root, timeout=10))\n\n\ndef load_page(root, slug, published=None):\n    source = published or root / "scripts/evals" / slug\n    data = published or root / "data/use-case-evals" / slug\n    config = json.loads((source / "eval.json").read_text())\n    items = read_rows(data / "fixture.jsonl")\n    if not items or any(not isinstance(i.get("item_id"), str) or not i["item_id"] or i.get("split") not in ("dev", "test") or not isinstance(i.get("text"), str) for i in items):\n        raise ValueError("fixture needs item_id, dev/test split and text")\n    if len({i["item_id"] for i in items}) != len(items):\n        raise ValueError("duplicate fixture ids")\n    for key, file in (("fixture_sha256", data / "fixture.jsonl"), ("recipe_sha256", source / "recipe.py"), ("checker_sha256", source / "check.py")):\n        if config.get(key) != sha256(file.read_bytes()):\n            raise ValueError("frozen " + key + " mismatch")\n    if config.get("runner_sha256") != sha256((source / "run_eval.py").read_bytes()):\n        raise ValueError("frozen runner_sha256 mismatch; generate runner before freezing eval.json")\n    if type(config.get("max_tokens")) is not int or config["max_tokens"] < 1 or not config.get("reasoning_effort"):\n        raise ValueError("freeze max_tokens and reasoning_effort")\n    if config.get("tool_choice", {}).get("type") != "function" or not config["tool_choice"].get("function", {}).get("name"):\n        raise ValueError("freeze a forced named tool_choice")\n    return source, data, config, items\n\n\ndef token_counter():\n    try:\n        import tiktoken\n    except ImportError:\n        raise ValueError("--plan requires tiktoken (python3 -m pip install tiktoken==0.14.0); no character-count fallback") from None\n    encoding = tiktoken.get_encoding("o200k_base")\n    return lambda text: len(encoding.encode(text, disallowed_special=()))\n\n\ndef make_plan(slug, config, items, recipe, arms, repeats, split, limit, rpm, pricing, count=None):\n    count = count or token_counter()\n    selected = [i for i in items if i["split"] == split][:limit]\n    dev = [i for i in items if i["split"] == "dev"][:limit]\n    if not selected:\n        raise ValueError("no items in selected split")\n    def budget(rows, n):\n        inputs = outputs = requests = 0\n        for item in rows:\n            # A loop recipe must supply all planned calls, including expected\n            # tool context. These are estimates, never measured API usage.\n            payloads = recipe.plan_requests(item, config) if hasattr(recipe, "plan_requests") else [recipe.request(item, config)]\n            if not payloads:\n                raise ValueError("recipe planning returned no requests")\n            for payload in payloads:\n                serialized = json.dumps({k: v for k, v in payload.items() if k != "model"}, ensure_ascii=False, sort_keys=True)\n                inputs += count(serialized)\n                outputs += config["max_tokens"]\n                requests += 1\n        return {"requests": requests * n, "input": inputs * n, "output": outputs * n}\n    scored = budget(selected, repeats)\n    tuning = budget(dev, config.get("tuning_repeats", 1))\n    total = {k: scored[k] * 2 + tuning[k] for k in scored}\n    weekly = pricing["personal"]["weeklyMillions"] * 1e6\n    price = next(p for p in pricing["prices"] if p["id"] == "gpt-6-1-sol")\n    n = len(selected)\n    # Paired bounded item differences in [-1,1], worst-case SD=1. Normal\n    # approximation, alpha=.05 two-sided and 80% power. Not a bootstrap result.\n    # Nonlinear set metrics require a dev SD on the metric\'s own scale.\n    metric = config["headline"]["metric"]\n    sd = config.get("planning", {}).get("paired_difference_sd")\n    assumption = "dev paired-difference SD on headline scale"\n    if sd is None and metric in ("accuracy", "mean", "count"):\n        bounds = config.get("planning", {}).get("score_bounds", [0, 1] if metric == "accuracy" else None)\n        if bounds and len(bounds) == 2 and bounds[1] > bounds[0]:\n            sd, assumption = bounds[1] - bounds[0], "worst-case bounded paired differences"\n    mde = None if sd is None else (1.959964 + 0.841621) * sd / math.sqrt(n)\n    if mde is not None and metric == "count":\n        mde *= n\n    return {"slug": slug, "split": split, "items": n, "repeats": repeats, "arms": arms,\n            "tokenizer": "o200k_base; serialized messages/tools and request overhead estimate",\n            "per_arm_scored": scored, "per_arm_tuning": tuning, "per_arm_rerun": scored,\n            "requests": total["requests"] * len(arms), "tokens": (total["input"] + total["output"]) * len(arms),\n            "output_assumption": "max_tokens per request including reasoning; cache savings not assumed; retries extra",\n            "personal_week_share_percent": (total["input"] + total["output"]) / weekly * 100 if "pareto" in arms else 0,\n            "personal_weekly_tokens": weekly, "rpm": rpm, "minutes": total["requests"] * len(arms) / rpm,\n            "baseline_cost_usd": (total["input"] * price["input_usd_per_mtok"] + total["output"] * price["output_usd_per_mtok"]) / 1e6 if "baseline" in arms else 0,\n            "minimum_detectable_difference": mde,\n            "mde_assumptions": assumption + "; approximate 95% confidence, 80% power, paired items not repeats; verdict also must exceed observed run ranges" if mde is not None else "not estimable for nonlinear headline without planning.paired_difference_sd from dev data; accept likely tie or supply SD"}\n\n\ndef checked_manifest(run, source, data):\n    manifest = json.loads((run / "manifest.json").read_text())\n    if manifest.get("format") != "uc-eval-run-v1":\n        raise ValueError("unknown run format")\n    for name, path in (("fixture.jsonl", data / "fixture.jsonl"), ("eval.json", source / "eval.json"), ("recipe.py", source / "recipe.py"), ("check.py", source / "check.py"), ("run_eval.py", source / "run_eval.py")):\n        if manifest["hashes"][name] != sha256(path.read_bytes()):\n            raise ValueError("run inputs changed: " + name)\n    if manifest.get("status") != "complete":\n        raise ValueError("run incomplete; inspect partial evidence before new generation")\n    if manifest["raw_sha256"] != sha256((run / "raw.jsonl.gz").read_bytes()):\n        raise ValueError("raw hash mismatch")\n    return manifest\n\n\ndef rescore(run, source, data, config, items):\n    if run.resolve().is_relative_to(data.resolve()) or run.resolve().is_relative_to(source.resolve()):\n        raise ValueError("rescore output must stay outside published evidence")\n    manifest = checked_manifest(run, source, data)\n    rows = read_rows(run / "raw.jsonl.gz")\n    selected = [i for i in items if i["split"] == manifest["split"]][:manifest["limit"]]\n    ids = {i["item_id"]: i for i in selected}\n    expected = {(i, a, r) for i in ids for a in manifest["arms"] for r in range(1, manifest["repeats"] + 1)}\n    seen = set()\n    if source.name == \'ai-unit-test-generation\' or config.get(\'sandbox\') == \'macos\':\n        from sandbox import score as sandbox_score\n        from types import SimpleNamespace\n        checker = SimpleNamespace(score=lambda item, arguments, config: sandbox_score(source / \'check.py\', item, arguments, config))\n    else:\n        checker = load_module(source / "check.py", "uc_checker")\n    scored = []\n    for row in rows:\n        approved_raw(row)\n        key = (row["item_id"], row["arm"], row["repeat"])\n        if key not in expected or key in seen:\n            raise ValueError("raw item/arm/repeat mismatch")\n        seen.add(key)\n        identity(row["arm"], row["request_model"], row["endpoint"], {"model": row["response_model"], "provider": row["provider_served"]})\n        if row["allow_fallbacks"] is not False or row["max_tokens"] != config["max_tokens"] or row["reasoning_effort"] != config["reasoning_effort"] or row["tool_choice"] != config["tool_choice"]:\n            raise ValueError("raw frozen request mismatch")\n        clean = {k: v for k, v in row.items() if k not in SCORE_KEYS}\n        clean.update(score_record(ids[row["item_id"]], row["tool_arguments"], checker, config))\n        scored.append(clean)\n    if seen != expected:\n        raise ValueError("raw missing items")\n    write_raw(run / "scored.jsonl.gz", scored)\n    return manifest, scored\n\n\ndef generate(slug, source, data, config, items, args, client_factory=None):\n    selected = [i for i in items if i["split"] == args.split][:args.limit]\n    if not selected:\n        raise ValueError("no items in selected split")\n    recipe = load_module(source / "recipe.py", "uc_recipe")\n    limiter = Limiter(args.rpm)\n    factory = client_factory or (lambda arm: Client(arm, "pareto" if arm == "pareto" else BASELINE_MODEL, PARETO_URL if arm == "pareto" else BASELINE_URL, api_key(arm), limiter))\n    clients = {arm: factory(arm) for arm in args.arms}\n    directory = Path(args.out).resolve() / slug\n    if directory.is_relative_to(data.resolve()) or directory.is_relative_to(source.resolve()) or \'data/use-case-evals\' in directory.as_posix():\n        raise ValueError("generation output must stay outside published evidence")\n    directory.mkdir(parents=True, exist_ok=True)\n    run = directory / datetime.now(timezone.utc).strftime("%Y-%m-%dT%H-%M-%S-%fZ")\n    run.mkdir()\n    manifest = {"format": "uc-eval-run-v1", "slug": slug, "status": "incomplete", "split": args.split,\n                "limit": args.limit, "arms": args.arms, "repeats": args.repeats,\n                "hashes": {name: sha256(path.read_bytes()) for name, path in (("fixture.jsonl", data / "fixture.jsonl"), ("eval.json", source / "eval.json"), ("recipe.py", source / "recipe.py"), ("check.py", source / "check.py"), ("run_eval.py", source / "run_eval.py"))}}\n    if getattr(args, \'freeze_sha\', None):\n        manifest[\'freeze_sha\'] = args.freeze_sha\n    (run / "manifest.json").write_bytes(json_bytes(manifest))\n    retain_runs(directory)\n    rows = []\n    try:\n        for repeat in range(1, args.repeats + 1):\n            for item in selected:\n                for arm in args.arms:\n                    rows.append(generate_record(item, arm, repeat, config, recipe, clients[arm]))\n                    # Persist each completed item so a timeout leaves evidence.\n                    write_raw(run / "raw.jsonl.gz", rows)\n        manifest["status"] = "complete"\n    finally:\n        if (run / "raw.jsonl.gz").exists():\n            manifest["raw_sha256"] = sha256((run / "raw.jsonl.gz").read_bytes())\n        (run / "manifest.json").write_bytes(json_bytes(manifest))\n    return run\n\n\ndef write_receipt(root, slug, source, data, run, manifest, rows, config, pending=False):\n    expected_arms = {\'pareto\'} if pending else {\'pareto\', \'baseline\'}\n    if manifest["split"] != "test" or set(manifest["arms"]) != expected_arms or manifest["limit"] is not None or manifest["repeats"] != config["repeats"] or config["repeats"] < (5 if slug == "llm-as-a-judge" else 3):\n        raise ValueError("receipt needs full paired frozen test repeats with no --limit")\n    git = lambda *args: subprocess.check_output(["git", *args], cwd=root, stderr=subprocess.PIPE, timeout=10)\n    freeze = manifest.get(\'freeze_sha\') or config.get("freeze_sha") or git("rev-parse", "HEAD").decode().strip()\n    git("merge-base", "--is-ancestor", freeze, "HEAD")\n    freeze_time = int(git("show", "-s", "--format=%ct", freeze))\n    for path in (source / "eval.json", source / "recipe.py", source / "check.py", source / "run_eval.py", data / "fixture.jsonl"):\n        if git("show", freeze + ":" + path.relative_to(root).as_posix()) != path.read_bytes():\n            raise ValueError("changed after freeze: " + path.name)\n    dates = {r["started_at"][:10] for r in rows}\n    if not dates or any(datetime.fromisoformat(r["started_at"].replace("Z", "+00:00")).timestamp() <= freeze_time for r in rows):\n        raise ValueError("receipt must follow freeze")\n    arms = {}\n    for arm in sorted(expected_arms):\n        subset = [r for r in rows if r["arm"] == arm]\n        models = {r["response_model"] for r in subset}\n        if len(models) != 1 or len({r["response_id"] for r in subset}) != len(subset) or any(r["errors"] for r in subset):\n            raise ValueError("run has errors, reused ids or changed model")\n        if not any(len({json.dumps(r["tool_arguments"], sort_keys=True) for r in subset if r["item_id"] == item}) > 1 for item in {r["item_id"] for r in subset}):\n            raise ValueError("repeat independence: identical outputs")\n        runs = []\n        for repeat in range(1, config["repeats"] + 1):\n            records = [r for r in subset if r["repeat"] == repeat]\n            runs.append({"repeat": repeat, "tokens": {k: sum(r["tokens"][k] for r in records) for k in TOKEN_KEYS},\n                         **{k: sum(r[k] for r in records) for k in ("errors", "invalid_tool_calls", "retries")},\n                         "truncations": sum(r["finish_reason"] == "length" for r in records),\n                         "response_ids_sha": sha256(json.dumps([r["response_id"] for r in records], ensure_ascii=False, separators=(",", ":")).encode())})\n        arms[arm] = {"model": subset[0]["request_model"], "endpoint": subset[0]["endpoint"], "max_tokens": config["max_tokens"], "runs": runs}\n        if arm == "pareto":\n            arms[arm]["response_model"] = subset[0]["response_model"]\n        else:\n            arms[arm].update(label="GPT-6.1 Sol", provider_served="OpenAI", reasoning_effort=config["reasoning_effort"])\n    fixture = read_rows(data / "fixture.jsonl")\n    prefix = data.relative_to(root).as_posix() + "/"\n    evaluation = {"schema": 2, "runner": (source / "run_eval.py").relative_to(root).as_posix(), "pareto_release": arms["pareto"]["response_model"],\n                  "fixture": {"path": prefix + "fixture.jsonl", "sha256": config["fixture_sha256"], "items": sum(i["split"] == "test" for i in fixture), "dev_items": sum(i["split"] == "dev" for i in fixture), "split": "test", **config["fixture_source"]},\n                  "prompt_sha256": sha256(config["prompt"].encode()), "recipe_sha256": config["recipe_sha256"], "eval_json_sha256": sha256((source / "eval.json").read_bytes()),\n                  "freeze_sha": freeze, **{k: config[k] for k in ("repeats", "seed", "unit", "aggregation", "headline", "secondary")}, "arms": arms, "variants": config.get("variants", {}),\n                  "raw": {"path": prefix + "runs/" + run.name + "/raw.jsonl.gz", "sha256": sha256((run / "scored.jsonl.gz").read_bytes())}}\n    if pending:\n        evaluation[\'baseline_status\'] = \'pending\'\n    receipt = {"date": max(dates), "result": "pass", "eval": evaluation}\n    candidate = run / "receipt.json"\n    candidate.write_bytes(json_bytes(receipt))\n    # Node owns schema/metric validation; validate before changing published data.\n    subprocess.run(["node", str(root / "scripts/evals/validate-receipt.mjs"), str(root), slug, str(candidate), str(run / "scored.jsonl.gz")], check=True, timeout=30, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n    target = root / evaluation["raw"]["path"]\n    target.parent.mkdir(parents=True, exist_ok=True)\n    target.write_bytes((run / "scored.jsonl.gz").read_bytes())\n    checks_path = root / "data/use-case-checks.json"\n    # Use JSON.stringify for byte identity with the existing builder/gate.\n    subprocess.run(["node", "-e", "const fs=require(\'fs\');const [p,s,r]=process.argv.slice(1);const d=JSON.parse(fs.readFileSync(p));const v=JSON.parse(fs.readFileSync(r));d.checks[s]=v;fs.writeFileSync(p,JSON.stringify(d,null,2)+\'\\\\n\');fs.writeFileSync(process.argv[4],JSON.stringify(v,null,2)+\'\\\\n\')", str(checks_path), slug, str(candidate), str(data / "receipt.json")], check=True, timeout=10)\n    # Keep only latest and previous published raw run, never other files.\n    previous = sorted(p for p in (data / "runs").iterdir() if p.is_dir() and not p.is_symlink())\n    for old in previous[:-2]:\n        if old != target.parent and {p.name for p in old.iterdir()} == {"raw.jsonl.gz"}:\n            (old / "raw.jsonl.gz").unlink()\n            old.rmdir()\n\n\ndef merge_baseline(root, slug, source, data, run, config, items):\n    receipt = json.loads((data / \'receipt.json\').read_text())\n    current = json.loads((root / \'data/use-case-checks.json\').read_text())[\'checks\'][slug]\n    if receipt != current or receipt[\'eval\'].get(\'baseline_status\') != \'pending\':\n        raise ValueError(\'merge requires the current pending receipt\')\n    e = receipt[\'eval\']\n    manifest, baseline = rescore(run, source, data, config, items)\n    if manifest[\'slug\'] != slug or manifest[\'arms\'] != [\'baseline\'] or manifest.get(\'freeze_sha\') != e[\'freeze_sha\']:\n        raise ValueError(\'baseline run must share the pending freeze sha\')\n    if manifest[\'split\'] != \'test\' or manifest[\'limit\'] is not None or manifest[\'repeats\'] != e[\'repeats\']:\n        raise ValueError(\'merge requires full frozen baseline repeats\')\n    for key, value in [(\'eval_json_sha256\', sha256((source / \'eval.json\').read_bytes())), (\'fixture\', config[\'fixture_sha256\'])]:\n        if (e[key][\'sha256\'] if key == \'fixture\' else e[key]) != value:\n            raise ValueError(\'pending frozen inputs changed\')\n    raw_path = (root / e[\'raw\'][\'path\']).resolve()\n    if not raw_path.is_relative_to(data.resolve()) or sha256(raw_path.read_bytes()) != e[\'raw\'][\'sha256\']:\n        raise ValueError(\'pending raw path/hash mismatch\')\n    pareto = read_rows(raw_path)\n    if any(r[\'arm\'] != \'pareto\' for r in pareto):\n        raise ValueError(\'pending raw contains another arm\')\n    # Revalidate the original receipt before combining, not just the new arm.\n    subprocess.run([\'node\', str(root / \'scripts/evals/validate-receipt.mjs\'), str(root), slug, str(data / \'receipt.json\'), str(raw_path)], check=True, timeout=30, stdout=subprocess.PIPE, stderr=subprocess.PIPE)\n    merged = dict(manifest, arms=[\'pareto\', \'baseline\'])\n    rows = pareto + baseline\n    write_raw(run / \'scored.jsonl.gz\', rows)\n    write_receipt(root, slug, source, data, run, merged, rows, config)\n\n\ndef main(root=None, published=None, pricing=None, slug_default=None):\n    root = Path(root).resolve() if root else None\n    parser = argparse.ArgumentParser(description="Two-phase use-case evals; no API calls in --plan or --rescore")\n    parser.add_argument("slug", nargs="?", default=slug_default)\n    parser.add_argument("--all", action="store_true")\n    modes = parser.add_mutually_exclusive_group()\n    modes.add_argument("--plan", action="store_true")\n    modes.add_argument("--generate", action="store_true")\n    modes.add_argument("--rescore", metavar="RUN")\n    modes.add_argument(\'--merge-baseline\', metavar=\'RUN\')\n    parser.add_argument(\'--pending\', action=\'store_true\')\n    parser.add_argument("--write-receipt", action="store_true")\n    parser.add_argument("--arms", help="default: both for planning, Pareto only for generation")\n    parser.add_argument("--repeats", type=int)\n    parser.add_argument("--split", choices=("dev", "test"), default="test")\n    parser.add_argument("--limit", type=int)\n    parser.add_argument("--rpm", type=float, default=18)\n    parser.add_argument("--out", default="/tmp/uc-smoke")\n    args = parser.parse_args()\n    if not (args.plan or args.generate or args.rescore or args.merge_baseline):\n        if args.write_receipt or args.arms:\n            args.generate = True\n        else:\n            parser.error(\'choose --plan, --generate, --rescore or --merge-baseline\')\n    if args.pending and (not args.write_receipt or args.plan or args.merge_baseline):\n        parser.error(\'--pending requires --write-receipt with generation or rescore\')\n    if args.merge_baseline and published:\n        parser.error(\'--merge-baseline requires the repository\')\n    args.arms = (args.arms or ("pareto,baseline" if args.plan else "pareto")).split(",")\n    if not args.arms or len(set(args.arms)) != len(args.arms) or set(args.arms) - {"pareto", "baseline"}:\n        parser.error("--arms must be pareto, baseline, or pareto,baseline")\n    if args.limit is not None and args.limit < 1 or args.repeats is not None and args.repeats < 1 or not math.isfinite(args.rpm) or not 0 < args.rpm <= 30:\n        parser.error("positive limit/repeats and rpm <= 30 required")\n    if args.all and (not args.plan or args.slug and not published):\n        parser.error("--all is only supported for --plan without a slug")\n    if args.write_receipt and (args.plan or published):\n        parser.error("--write-receipt requires a repository generation or rescore")\n    if published and args.slug != slug_default:\n        parser.error("this download belongs to " + slug_default)\n    slugs = [args.slug] if args.slug else []\n    if args.all:\n        slugs = [slug_default] if published else sorted(p.parent.name for p in (root / "scripts/evals").glob("*/eval.json"))\n        if not slugs:\n            print("No eval pages yet. requests=0 tokens=0 baseline_cost_usd=0 minimum_detectable_difference=n/a")\n            return\n    if not slugs or any(not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", s) for s in slugs):\n        parser.error("provide a safe slug or --all --plan")\n    for slug in slugs:\n        source, data, config, items = load_page(root, slug, published)\n        repeats = args.repeats or config["repeats"]\n        if args.plan:\n            recipe = load_module(source / "recipe.py", "uc_recipe")\n            print(json.dumps(make_plan(slug, config, items, recipe, args.arms, repeats, args.split, args.limit, args.rpm, pricing or canon(root)), indent=2))\n            continue\n        args.repeats = repeats\n        if args.merge_baseline:\n            merge_baseline(root, slug, source, data, Path(args.merge_baseline).resolve(), config, items)\n            print(\'Merged baseline; rebuild pages and run check-use-case-evals\')\n            continue\n        if args.pending and args.generate and args.arms != [\'pareto\']:\n            parser.error(\'--pending generates Pareto only\')\n        if root:\n            pending_path = data / \'receipt.json\'\n            prior = json.loads(pending_path.read_text()) if pending_path.exists() else {}\n            args.freeze_sha = prior.get(\'eval\', {}).get(\'freeze_sha\') if prior.get(\'eval\', {}).get(\'baseline_status\') == \'pending\' else None\n            args.freeze_sha = args.freeze_sha or config.get(\'freeze_sha\') or subprocess.check_output([\'git\', \'rev-parse\', \'HEAD\'], cwd=root, timeout=10).decode().strip()\n        if args.generate:\n            run = generate(slug, source, data, config, items, args)\n            print("Generated unscored run: " + str(run))\n            if not args.write_receipt:\n                continue\n        else:\n            run = Path(args.rescore).resolve()\n        manifest, rows = rescore(run, source, data, config, items)\n        print("Rescored without API calls: " + str(run / "scored.jsonl.gz"))\n        if args.write_receipt:\n            write_receipt(root, slug, source, data, run, manifest, rows, config, pending=args.pending)\n            print("Wrote receipt; rebuild pages and run check-use-case-evals before committing")\n\n\ndef entry(**kwargs):\n    try:\n        main(**kwargs)\n    except Exception as error:\n        # Provider bodies, recipe exception text and environment values can\n        # contain keys. CLI failures disclose only their class, not payloads.\n        print("Eval stopped: " + type(error).__name__ + "; check configuration, hashes, dependencies and partial run artifacts", file=sys.stderr)\n        return 1\n    return 0\n', "uc_app", "exec"), app.__dict__)
if __name__ == "__main__":
    sys.exit(app.entry(published=Path(__file__).resolve().parent, pricing={'personal': {'name': 'Personal', 'monthly': 10, 'weeklyMillions': 50, 'rpm': 20, 'campaign': 'pricing_personal', 'label': 'Start Personal'}, 'prices': [{'id': 'claude-fable-5-1', 'name': 'Claude Fable 5.1', 'provider': 'Anthropic', 'tier': 'premium', 'input_usd_per_mtok': 10, 'output_usd_per_mtok': 50, 'cache_read_multiplier': 0.025, 'cache_read_usd_per_mtok': 0.25, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27', 'note': 'Same rate card as Fable 5, but cache reads bill at 0.025x base input ($0.25/MTok) instead of the standard 0.1x. Uses the newer tokenizer.'}, {'id': 'claude-fable-5', 'name': 'Claude Fable 5', 'provider': 'Anthropic', 'tier': 'premium', 'input_usd_per_mtok': 10, 'output_usd_per_mtok': 50, 'cache_read_multiplier': 0.1, 'cache_read_usd_per_mtok': 1, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27'}, {'id': 'claude-opus-5-5', 'name': 'Claude Opus 5.5', 'provider': 'Anthropic', 'tier': 'frontier', 'input_usd_per_mtok': 4, 'output_usd_per_mtok': 20, 'cache_read_multiplier': 0.05, 'cache_read_usd_per_mtok': 0.2, 'cache_write_usd_per_mtok': 5, 'cache_write_multiplier': 1.25, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27', 'note': "Verified 2026-10-27 on Anthropic's pricing page: $4 input / $20 output per MTok, 5-minute cache writes $5, 1-hour cache writes $8, cache reads $0.20 (0.05x input), batch $2 / $10. Full 1M context at standard pricing."}, {'id': 'claude-opus-5', 'name': 'Claude Opus 5', 'provider': 'Anthropic', 'tier': 'frontier', 'input_usd_per_mtok': 5, 'output_usd_per_mtok': 25, 'cache_read_multiplier': 0.1, 'cache_read_usd_per_mtok': 0.5, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27'}, {'id': 'claude-opus-4-8', 'name': 'Claude Opus 4.8', 'provider': 'Anthropic', 'tier': 'frontier-prior', 'input_usd_per_mtok': 5, 'output_usd_per_mtok': 25, 'cache_read_multiplier': 0.1, 'cache_read_usd_per_mtok': 0.5, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27'}, {'id': 'claude-sonnet-5-5', 'name': 'Claude Sonnet 5.5', 'provider': 'Anthropic', 'tier': 'mid', 'input_usd_per_mtok': 2, 'output_usd_per_mtok': 10, 'cache_read_multiplier': 0.05, 'cache_read_usd_per_mtok': 0.1, 'cache_write_usd_per_mtok': 2.5, 'cache_write_multiplier': 1.25, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27', 'note': 'Verified 2026-10-27: $2 input / $10 output per MTok, cache reads $0.10 (0.05x input), 5-minute cache writes $2.50, batch 50% off. API model id claude-sonnet-5-5.'}, {'id': 'claude-sonnet-5', 'name': 'Claude Sonnet 5', 'provider': 'Anthropic', 'tier': 'mid', 'input_usd_per_mtok': 2, 'output_usd_per_mtok': 10, 'cache_read_multiplier': 0.1, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27', 'note': '$2/$10 is now the standard rate. The increase to $3/$15 previously scheduled for 2026-09-01 was cancelled; Anthropic made the introductory price the standard price (platform release notes, August 10, 2026: https://platform.claude.com/docs/en/release-notes/overview).'}, {'id': 'claude-haiku-5-5', 'name': 'Claude Haiku 5.5', 'provider': 'Anthropic', 'tier': 'volume', 'input_usd_per_mtok': 0.1, 'output_usd_per_mtok': 0.5, 'cache_read_multiplier': 0.1, 'cache_read_usd_per_mtok': 0.01, 'cache_write_usd_per_mtok': 0.125, 'cache_write_multiplier': 1.25, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27', 'note': "Verified 2026-10-27 on Anthropic's pricing page: $0.10 input / $0.50 output per MTok for prompts up to 100,000 tokens; $0.50 / $2.50 for prompts over 100,000 tokens. 5-minute cache writes $0.125, 1-hour $0.20, cache reads $0.01. Batch $0.05 / $0.25 (up to 100K)."}, {'id': 'claude-haiku-4-5', 'name': 'Claude Haiku 4.5', 'provider': 'Anthropic', 'tier': 'volume', 'input_usd_per_mtok': 1, 'output_usd_per_mtok': 5, 'cache_read_multiplier': 0.1, 'cache_read_usd_per_mtok': 0.1, 'batch_discount': 0.5, 'source': 'https://platform.claude.com/docs/en/about-claude/pricing', 'verified': '2026-10-27'}, {'id': 'gpt-6-1-sol', 'name': 'GPT-6.1 Sol', 'provider': 'OpenAI', 'tier': 'frontier', 'input_usd_per_mtok': 2, 'output_usd_per_mtok': 10, 'cache_read_usd_per_mtok': 0.1, 'cache_read_multiplier': 0.05, 'source': 'https://developers.openai.com/api/docs/pricing', 'verified': '2026-10-27', 'note': 'Verified 2026-10-27 against OpenAI Standard pricing, short context: $2 input / $0.10 cached input / $10 output per MTok. API model id gpt-6.1-sol.'}, {'id': 'gpt-6-luna', 'name': 'GPT-6 Luna', 'provider': 'OpenAI', 'tier': 'volume', 'input_usd_per_mtok': 0.1, 'output_usd_per_mtok': 0.5, 'cache_read_usd_per_mtok': 0.01, 'cache_read_multiplier': 0.1, 'source': 'https://developers.openai.com/api/docs/pricing', 'verified': '2026-10-27', 'note': 'Verified 2026-10-27 against OpenAI Standard pricing, short context: $0.10 input / $0.01 cached input / $0.50 output per MTok. API model id gpt-6-luna.'}, {'id': 'pareto', 'name': 'Pareto', 'provider': 'Unbiased', 'tier': 'blended', 'source': 'https://unbiased.ai/pricing/', 'verified': '2026-10-27', 'note': 'Published Pareto rates. Blended model: several LLMs per request, one synthesized answer. Benchmark receipts on the model card.', 'input_usd_per_mtok': 0.8, 'output_usd_per_mtok': 3.2, 'cache_read_usd_per_mtok': 0.03, 'cache_read_multiplier': 0.0375}, {'id': 'kimi-k3', 'name': 'Kimi K3', 'provider': 'Moonshot AI', 'tier': 'mid', 'input_usd_per_mtok': 3, 'output_usd_per_mtok': 15, 'cache_read_multiplier': 0.1, 'source': 'https://platform.kimi.ai/docs/pricing/chat', 'verified': '2026-10-27'}]}, slug_default='llm-as-a-judge'))