Hot dog.Not hot dog.

# /// script
# requires-python = ">=3.10"
# dependencies = ["httpx", "duckdb==1.5.6"]
# ///
"""hotdogbench: ask vision models "is this a hot dog?" through OpenRouter and score the answers.

Input is a table of photos with columns id, image (JPEG bytes), is_hot_dog, category:
  (default)                                   the public dataset, read straight from S3
  --input dist/hotdogbench.parquet            a local copy (or any parquet with those columns)
  --input local --per-group 25                a balanced sample of downloaded photos (building the dataset)
Every model gets the same photo bytes and the same question, one photo per request, and answers as JSON
{"hot_dog": true|false}. Requests come off one queue with an adaptive in-flight limit: it starts at
--start-concurrency, doubles until OpenRouter first pushes back (429/5xx), then cuts to 70% and climbs by one
per round trip, so each model runs just under its rate cap. Throttled photos go back on the queue.
Results go to ./out/<run>/ (in the folder you run it from):
  <model>.parquet   one row per photo: answer, latency, tokens, cost (from OpenRouter usage), retries, error
  summary.json      per model: accuracy (overall and by category), unanswered, wall time, latency, cost
Usage:
  export OPENROUTER_API_KEY=...   (or put it in ./.env)
  uv run https://hotdogbench.com/hotdogbench.py --models cheap    # Jev, Luna, Haiku, Sol: about $2
  uv run https://hotdogbench.com/hotdogbench.py --models all      # all seven: about $21
"""
import argparse
import asyncio
import base64
import hashlib
import json
import os
import random
import statistics
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path

import duckdb
import httpx

HERE = Path(__file__).resolve().parent
DATA_URL = "https://md-ecosystem-public.s3.amazonaws.com/hotdogbench.parquet"  # the public dataset
URL = "https://openrouter.ai/api/v1/chat/completions"
# Jev answers typed questions through OpenRouter's Decisions API; images go in `state` as image_url parts.
DECISIONS_URL = "https://openrouter.ai/api/alpha/decisions"
QUESTION = "Is the food in this photo a hot dog? Answer with JSON: {\"hot_dog\": true} or {\"hot_dog\": false}."
SCHEMA = {
    "type": "json_schema",
    "json_schema": {
        "name": "verdict",
        "strict": True,
        "schema": {"type": "object", "properties": {"hot_dog": {"type": "boolean"}},
                   "required": ["hot_dog"], "additionalProperties": False},
    },
}
# OpenRouter id -> extra request params. Lowest reasoning each model allows; temperature 0 where accepted.
MODELS = {
    "openai/gpt-6.1-sol": {"reasoning": {"effort": "minimal"}},
    "openai/gpt-6-luna": {"reasoning": {"effort": "minimal"}},
    "openai/gpt-6-astra": {"reasoning": {"effort": "minimal"}},
    "typesafe/jev-1.14": {"decisions": True},
    # Decisions API. Pinned to Cloudflare's own endpoint: DeepInfra's rejects image_url parts.
    # (microsoft/microsoft-decision-1 rejects image_url parts on every provider, so it can't take this test.)
    "cloudflare/clef": {"decisions": True, "provider": {"only": ["Cloudflare"], "allow_fallbacks": False}},
    "openai/gpt-6-luna-decisions": {"decisions": True},
    "perplexity/pplx-decider-v1.1-27b": {"decisions": True},
    "anthropic/claude-opus-5.5": {"temperature": 0},
    "anthropic/claude-sonnet-5.5": {"temperature": 0},
    "anthropic/claude-haiku-5.5": {"temperature": 0},
}
RETRY = {408, 409, 425, 429, 500, 502, 503, 504, 520, 522, 524, 529}


@dataclass
class Result:
    id: int
    open_images_id: str | None
    category: str
    is_hot_dog: bool
    scored: bool = True
    pred: bool | None = None
    raw: str | None = None
    latency_s: float | None = None
    prompt_tokens: int | None = None
    completion_tokens: int | None = None
    reasoning_tokens: int | None = None
    cost_usd: float | None = None
    provider: str | None = None
    retries: list[str] = field(default_factory=list)
    error: str | None = None


def load_photos(args) -> list[dict]:
    con = duckdb.connect()
    if args.input == "local":
        # usable downloads so far: saved, and Flickr page still CC BY 2.0
        rows = con.sql(f"""
            WITH s AS (SELECT *, row_number() OVER () AS rn FROM read_csv('{HERE / "fetch_status.csv"}', all_varchar=true)),
            last AS (SELECT * FROM s QUALIFY row_number() OVER (PARTITION BY ImageID ORDER BY rn DESC) = 1)
            SELECT ImageID, grp FROM last WHERE status = 'ok' AND page_license = 'by/2.0'
            QUALIFY row_number() OVER (PARTITION BY grp ORDER BY hash(ImageID || 'sample')) <= {args.per_group}
        """).fetchall()
        photos = [{"id": i, "image": (HERE / "images" / f"{image_id}.jpg").read_bytes(), "category": grp,
                   "is_hot_dog": grp == "hot_dog", "scored": True, "open_images_id": image_id}
                  # stable order (Python's hash() is randomized per process, which scrambled ids between runs)
                  for i, (image_id, grp) in enumerate(sorted(rows, key=lambda r: hashlib.md5(r[0].encode()).hexdigest()), 1)]
        bonus = HERE / "extras" / "magritte_hot_dog.jpg"
        if bonus.exists():  # the unscored bonus row, as in the built dataset
            photos.append({"id": 1001, "image": bonus.read_bytes(), "category": "magritte", "is_hot_dog": True, "scored": False})
        return photos
    rel = con.sql(f"SELECT * FROM read_parquet('{args.input}', union_by_name = true) ORDER BY id")
    return [dict(zip(rel.columns, r)) for r in rel.fetchall()]


def body_for(model: str, image: bytes) -> dict:
    url = "data:image/jpeg;base64," + base64.b64encode(image).decode()
    if MODELS[model].get("decisions"):
        extra = {k: v for k, v in MODELS[model].items() if k != "decisions"}  # e.g. provider routing
        return {"model": model,
                "state": ["Photo:", {"type": "image_url", "image_url": {"url": url}}],
                "questions": {"hot_dog": {"type": "noul", "instructions": "Is the food in the photo a hot dog?"}},
                **extra}
    body = {
        "model": model,
        "messages": [{"role": "user", "content": [
            {"type": "image_url", "image_url": {"url": url}},
            {"type": "text", "text": QUESTION},
        ]}],
        "max_tokens": 1024,
        "response_format": SCHEMA,
        "usage": {"include": True},
        **MODELS[model],
    }
    return body


def parse(text: str) -> bool | None:
    try:
        v = json.loads(text.strip().removeprefix("```json").removesuffix("```").strip())["hot_dog"]
        return v if isinstance(v, bool) else None
    except (ValueError, KeyError, TypeError, AttributeError):
        low = (text or "").lower()
        if '"hot_dog": true' in low or '"hot_dog":true' in low:
            return True
        if '"hot_dog": false' in low or '"hot_dog":false' in low:
            return False
        return None


class Throttle(Exception):
    """OpenRouter pushed back (429, 5xx, or an upstream rate-limit error); the photo goes back on the queue."""
    def __init__(self, tag: str, retry_after: float | None = None, cut: bool = True):
        super().__init__(tag)
        self.tag, self.retry_after, self.cut = tag, retry_after, cut  # cut=False: a dropped connection, not a rate signal


class Limiter:
    """AIMD concurrency limit. Slow start (double per window) until the first throttle, then +1 per window of
    `limit` successes; on a throttle cut to 70%, at most once per cooldown so a burst of 429s counts once."""
    def __init__(self, start: int, cap: int):
        self.limit, self.cap, self.in_flight = float(start), cap, 0
        self.slow_start, self.successes, self.throttles, self.transport_errors = True, 0, 0, 0
        self.last_cut, self.cond = 0.0, asyncio.Condition()
        self.trace: list[tuple[float, int]] = []  # (seconds since start, limit) on every change
        self.t0 = time.perf_counter()

    def _mark(self):
        self.trace.append((round(time.perf_counter() - self.t0, 2), int(self.limit)))

    async def acquire(self):
        async with self.cond:
            await self.cond.wait_for(lambda: self.in_flight < int(self.limit))
            self.in_flight += 1

    async def release(self, ok: bool | None, latency: float | None = None):
        """ok=True success, False throttle (cut), None neutral (dropped connection: retried, limit unchanged)."""
        async with self.cond:
            self.in_flight -= 1
            now = time.perf_counter()
            if ok is None:
                self.transport_errors += 1
            elif ok:
                self.successes += 1
                if self.successes >= int(self.limit):  # one window of successes
                    self.successes = 0
                    self.limit = min(self.cap, self.limit * 2 if self.slow_start else self.limit + 1)
                    self._mark()
            else:
                self.throttles += 1
                if now - self.last_cut > max(1.0, latency or 1.0):  # one cut per round trip
                    self.slow_start, self.successes, self.last_cut = False, 0, now
                    self.limit = max(1.0, self.limit * 0.7)
                    self._mark()
            self.cond.notify_all()


async def ask_once(client, key, model, body, res: Result) -> None:
    """One request. Fills `res` on an answer or a final error; raises Throttle when it should be retried."""
    t0 = time.perf_counter()
    try:
        r = await client.post(DECISIONS_URL if "questions" in body else URL, json=body,
                              headers={"Authorization": f"Bearer {key}"})
    except (httpx.HTTPError, OSError) as e:  # stale keep-alive connections show up as SSLError/ReadError
        raise Throttle(type(e).__name__, retry_after=0.0, cut=False)
    res.latency_s = time.perf_counter() - t0
    if r.status_code in RETRY:
        ra = r.headers.get("retry-after")
        raise Throttle(str(r.status_code), float(ra) if ra and ra.replace(".", "").isdigit() else None)
    try:
        data = r.json()
    except ValueError:
        raise Throttle(f"{r.status_code}/badjson")
    if r.status_code != 200 or "error" in data:
        err = data.get("error") or {}
        code = err.get("code") if isinstance(err, dict) else None
        reason = ((err.get("metadata") or {}).get("reason") if isinstance(err, dict) else None) or ""
        if r.status_code == 402 and "in_flight" in reason:  # credit reservations for in-flight requests: back off
            raise Throttle("402/in_flight", retry_after=2.0)
        if code in RETRY:
            raise Throttle(f"200/err{code}")
        res.error = f"{r.status_code}: {json.dumps(err)[:300]}"
        return
    u = data.get("usage") or {}
    if "answers" in data:  # Decisions API: noul probability that it's a hot dog
        p = (data["answers"].get("hot_dog") or {}).get("noul")
        res.raw, res.pred = str(p), (None if p is None else p >= 0.5)
        res.prompt_tokens, res.completion_tokens = u.get("input_tokens"), u.get("output_tokens")
        res.cost_usd, res.provider = u.get("cost"), data.get("provider")
        return
    msg = data["choices"][0]["message"]
    res.raw = (msg.get("content") or "")[:500]
    res.pred = parse(res.raw)
    res.prompt_tokens, res.completion_tokens = u.get("prompt_tokens"), u.get("completion_tokens")
    res.reasoning_tokens = (u.get("completion_tokens_details") or {}).get("reasoning_tokens")
    res.cost_usd, res.provider = u.get("cost"), data.get("provider")


MAX_ATTEMPTS = 8


def summarize(model: str, all_results: list[Result], wall: float) -> dict:
    bonus = {str(r.id): r.pred for r in all_results if not r.scored}  # e.g. the Magritte painting
    results = [r for r in all_results if r.scored]
    answered = [r for r in results if r.pred is not None]
    def acc(rs):
        rs = [r for r in rs if r.pred is not None]
        return round(sum(r.pred == r.is_hot_dog for r in rs) / len(rs), 4) if rs else None
    lat = sorted(r.latency_s for r in results if r.latency_s is not None and r.error is None)
    cost = sum(r.cost_usd or 0 for r in results)
    pos = [r for r in answered if r.is_hot_dog]
    neg = [r for r in answered if not r.is_hot_dog]
    tpr = sum(r.pred for r in pos) / len(pos) if pos else None
    tnr = sum(not r.pred for r in neg) / len(neg) if neg else None
    return {
        "model": model,
        "photos": len(results),
        "answered": len(answered),
        "unanswered": len(results) - len(answered),
        "errors": sum(r.error is not None for r in results),
        "accuracy": acc(results),
        "balanced_accuracy": round((tpr + tnr) / 2, 4) if tpr is not None and tnr is not None else None,
        "accuracy_by_category": {c: acc([r for r in results if r.category == c])
                                 for c in sorted({r.category for r in results})},
        "false_alarms": sum(1 for r in neg if r.pred),
        "missed_hot_dogs": sum(1 for r in pos if not r.pred),
        "wall_s": round(wall, 2),
        "photos_per_s": round(len(all_results) / wall, 2) if wall else None,
        "latency_p50_s": round(statistics.median(lat), 3) if lat else None,
        "latency_p95_s": round(lat[int(0.95 * (len(lat) - 1))], 3) if lat else None,
        "cost_usd": round(cost, 5),
        "cost_per_1000_photos_usd": round(cost / len(results) * 1000, 3) if results else None,
        "prompt_tokens_per_photo": round(sum(r.prompt_tokens or 0 for r in results) / max(1, len(answered)), 1),
        "reasoning_tokens": sum(r.reasoning_tokens or 0 for r in results),
        "requests_retried": sum(1 for r in results if r.retries),
        "providers": sorted({r.provider for r in results if r.provider}),
        "unscored_answers": bonus,  # id -> said hot dog?
    }


async def run_model(model, photos, key, start, cap, out_dir) -> dict:
    """One queue of photos, `cap` workers, and an AIMD limiter deciding how many requests are in flight."""
    lim = Limiter(start, cap)
    queue: asyncio.Queue = asyncio.Queue()
    results: dict[int, Result] = {}
    for p in photos:
        res = Result(id=int(p["id"]), open_images_id=p.get("open_images_id"), category=p["category"],
                     is_hot_dog=bool(p["is_hot_dog"]), scored=p.get("scored") is not False)
        results[res.id] = res
        queue.put_nowait((0.0, p, body_for(model, p["image"]), res))
    pending = len(photos)
    done_evt = asyncio.Event()
    peak_in_flight = 0

    async def worker(client):
        nonlocal pending, peak_in_flight
        while not done_evt.is_set():
            try:
                not_before, p, body, res = await asyncio.wait_for(queue.get(), timeout=0.5)
            except asyncio.TimeoutError:
                continue
            wait = not_before - time.perf_counter()
            if wait > 0:
                await asyncio.sleep(wait)
            await lim.acquire()
            peak_in_flight = max(peak_in_flight, lim.in_flight)
            try:
                await ask_once(client, key, model, body, res)
                await lim.release(True)
            except Throttle as t:
                await lim.release(False if t.cut else None, res.latency_s)
                res.retries.append(t.tag)
                if len(res.retries) >= MAX_ATTEMPTS:
                    res.error = "retries exhausted: " + ",".join(res.retries[-3:])
                else:  # back on the queue; honour Retry-After, else a short jittered pause
                    delay = t.retry_after if t.retry_after is not None else min(20, 0.5 * 2 ** len(res.retries)) * random.random()
                    queue.put_nowait((time.perf_counter() + delay, p, body, res))
                    continue
            pending -= 1
            if pending == 0:
                done_evt.set()

    async def progress(t0):
        while not done_evt.is_set():
            await asyncio.sleep(5)
            n = len(photos) - pending
            print(f"  {model}: {n}/{len(photos)} done, limit {int(lim.limit)}, in flight {lim.in_flight}, "
                  f"throttles {lim.throttles}, dropped conns {lim.transport_errors}, {n / (time.perf_counter() - t0):.1f}/s", flush=True)

    limits = httpx.Limits(max_connections=cap, max_keepalive_connections=cap)
    async with httpx.AsyncClient(timeout=httpx.Timeout(120), limits=limits) as client:
        t0 = time.perf_counter()
        lim.t0 = t0
        tasks = [asyncio.create_task(worker(client)) for _ in range(cap)] + [asyncio.create_task(progress(t0))]
        await done_evt.wait()
        wall = time.perf_counter() - t0
        for t in tasks:
            t.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)
    ordered = [results[int(p["id"])] for p in photos]
    con = duckdb.connect()
    rows = [asdict(r) for r in ordered]
    for r in rows:
        r["retries"] = ",".join(r["retries"])
    con.execute("CREATE TABLE t AS SELECT * FROM (SELECT unnest(?, recursive := true))", [rows])
    con.execute(f"COPY t TO '{out_dir / (model.replace('/', '__') + '.parquet')}'")
    s = summarize(model, ordered, wall)
    s["concurrency"] = {"start": start, "cap": cap, "final_limit": int(lim.limit), "peak_in_flight": peak_in_flight,
                        "throttles": lim.throttles, "transport_errors": lim.transport_errors,
                        "limit_trace": lim.trace}
    return s


def openrouter_key() -> str:
    key = os.environ.get("OPENROUTER_API_KEY") or os.environ.get("OPENROUTER_TOKEN")
    for env in (Path.cwd() / ".env", HERE.parent / ".env"):
        if key or not env.exists():
            continue
        for line in env.read_text().splitlines():
            k, _, v = line.partition("=")
            if k.strip() in ("OPENROUTER_API_KEY", "OPENROUTER_TOKEN"):
                key = v.strip().strip('"').strip("'")
    if not key:
        raise SystemExit("Set OPENROUTER_API_KEY")
    return key


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    local = HERE / "dist" / "hotdogbench.parquet"
    ap.add_argument("--input", default=str(local) if local.exists() else DATA_URL)
    ap.add_argument("--per-group", type=int, default=25, help="with --input local: photos per category")
    ap.add_argument("--models", default="all", help="'all', 'cheap' (Jev, Luna, Haiku, Sol) or comma-separated OpenRouter ids")
    ap.add_argument("--start-concurrency", type=int, default=8, help="in-flight requests to start from")
    ap.add_argument("--max-concurrency", type=int, default=256, help="ceiling for the adaptive limit")
    ap.add_argument("--run", default=time.strftime("run_%Y%m%d_%H%M%S"))
    ap.add_argument("--out", default=str(Path.cwd() / "out"), help="results folder (default ./out)")
    args = ap.parse_args()

    cheap = ["typesafe/jev-1.14", "openai/gpt-6-luna", "anthropic/claude-haiku-5.5", "openai/gpt-6.1-sol"]
    models = {"all": list(MODELS), "cheap": cheap}.get(args.models) or args.models.split(",")
    photos = load_photos(args)
    out_dir = Path(args.out) / args.run
    out_dir.mkdir(parents=True, exist_ok=True)
    key = openrouter_key()
    print(f"{len(photos)} photos x {len(models)} models -> {out_dir}")
    summaries = []
    for m in models:  # models run one after another so their timings don't interfere
        s = asyncio.run(run_model(m, photos, key, args.start_concurrency, args.max_concurrency, out_dir))
        summaries.append(s)
        line = {k: s[k] for k in ("model", "accuracy", "unanswered", "errors", "wall_s",
                                  "photos_per_s", "latency_p50_s", "cost_per_1000_photos_usd")}
        line.update({k: s["concurrency"][k] for k in ("final_limit", "peak_in_flight", "throttles", "transport_errors")})
        print(json.dumps(line), flush=True)
        (out_dir / "summary.json").write_text(json.dumps(summaries, indent=2))


if __name__ == "__main__":
    main()